Fix Codex limits, Claude counting, and agent usage reliability from community PRs (#14049)

* Read Codex app-server replies from the raw fd (#13703)

* Resolve Codex through mise which instead of running the lazy launcher (#13109)

* Skip the Codex app-server probe when there are no credentials (#13106)

Adapted: credentials are checked in the home being probed rather than in
the CODEX_HOME environment variable, since each registered account is
probed in its own home, so a signed-out secondary account isn't hidden
behind the primary's login. A home without credentials reports "Waiting
for auth" like any other signed-out home. The credentials store setting is
read with tomllib, so a single-quoted value counts too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Show the Codex CLI's own error when its app-server dies (#8977)

Detect an app-server that exits or stops answering, and report the end of
its stderr instead of a bare RPC method name. Rebased onto the raw-fd
reply reader; the switch from "-a on-request" to "-a never" is left out,
keeping the current approval flags.

* Count pi sessions when HOME is a git checkout (#13209)

* Count only OpenAI-backed native sessions as Codex usage (#12032)

* Deduplicate Pi usage across forked sessions (#8602)

* Skip unchanged native Codex token snapshots (#10531)

* Count omp and pi profile sessions in the agent usage collectors (#9546)

`omp --profile=<name>` (and pi's equivalent) relocates the whole agent
tree under <base>/profiles/<name>/. The Claude and Codex collectors only
ever scanned <base>/agent/sessions, so a subscription driven entirely
through a profile was invisible to the agents panel: no tokens by day, no
tokens by model, no prompt or session counts.

Discover the profile roots alongside the default one. Sessions are keyed
by file path, so a profile adds sessions instead of double-counting the
default root, and a missing or unreadable profiles directory leaves the
existing behavior untouched.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014zFbJcDEEpV5BAmsH6kAB3

* Skip unrelated Codex session lines before JSON parsing (#12803)

Adapted: session_meta lines also pass the pre-filter, since the provider
filter from #12032 reads them to skip rollouts served by a non-OpenAI
provider.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Read only the Codex session files that changed since the last scan (#12595)

Native Codex rollouts keep per-file totals between runs, replayed while a
file's mtime and size are unchanged. Rebased onto the session_meta
provider filter, snapshot dedup, and line pre-filter, which now live in
the per-file reader. pi and omp sessions are left out of the per-file
cache: a forked pi session repeats its parent's messages, so they are
deduplicated across the whole tree on every scan.

* Count streamed Claude messages by their highest-output usage line (#10606)

Claude Code writes a streamed assistant response as several transcript
lines that share one message id, one per content block. Each line
carries a usage object. The first line's output_tokens is a placeholder,
often 1, and the last line has the real count. Input and cache fields
usually match across the lines.

The scanner dedupes by message id and keeps the first line it sees, so
it under-counts output tokens. On a machine with 2,577 transcripts it
reported 39.0M output tokens against 60.1M used, a 35% shortfall. Input
and both cache fields differed by under 0.01%.

Keep the line with the highest output count, with the last one scanned
winning a tie. The whole line is kept because a response can fall back
to another model mid-stream. Those lines are separate snapshots with
different cache figures and a different model, and taking a maximum per
field across them over-counts cache tokens and credits the wrong model.

The zero-usage check now runs before dedup, so a zero-usage first line
no longer claims a message id and hides a later line with real usage.

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
Co-authored-by: GPT-6 Astra <noreply@openai.com>

* Index Claude transcripts so the agents refresh reads only what was appended (#8313)

omarchy-agent-usage-claude re-parsed every line of every transcript under
~/.claude/projects on each refresh: no mtime cutoff, no memory of the last
pass. The agents widget is on by default and ticks every 15 minutes, so the
cost grew for the life of the machine. After one month here that was 803
files, 640 MB, 127k lines and 57k JSON parses per tick, about 1 core-second,
pushed through the page cache every quarter hour forever.

Keep a per-file index next to the scan cache: the unique usage records
already parsed out of each transcript and the byte offset they end at. A
file whose size and mtime match is not opened; a file that grew is read
from the stored offset; a file that shrank or was rewritten is read from
the start. --force drops the index and rescans from scratch.

The summary is built from the indexed records in the same directory order
the walk always used. That matters: when a resumed session carries earlier
messages, the same message id appears in two files with different usage,
and the first file visited wins. 91 ids differed on this machine; sorting
the walk moved one model's output total by 25k tokens. Output is now
byte-identical to the previous scan on a frozen copy of the corpus, cold,
warm, and after an append.

Warm refresh: 1.0 s -> 0.10 s of CPU, of which the scan itself is 70 ms;
the index for this corpus is 2.9 MB.

Adapted:
- Rebased onto #10606: the highest-output rule for streamed messages now
  lives where the index parses records, and decides between files too.
- The index records the timezone it was written in, and a change rereads
  every transcript, since its records hold local days.
- A file only counts as appended to when its inode and the hash of what
  was already read still match, so a transcript replaced by a larger one,
  or rewritten in place, is read from the start.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Label a Claude Team seat by its subscription, not its rate-limit tier (#11109)

The collector built the plan label from the OAuth rateLimitTier first, so a
Team premium seat, which runs on default_claude_max_5x, showed in the agents
panel as "Max 5x". Lead with subscriptionType and keep the multiplier as its
qualifier: Max still reads "Max 5x", a Team seat reads "Team 5x".

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

* Label the Claude plan from the profile the CLI refreshes (#7225)

Adapted: the profile is found the same way current_account_id() finds it,
now shared as profile_path(): ~/.claude.json for the default home, the
home's own .claude.json otherwise. The original fell back to ~/.claude.json
for any home without CLAUDE_CONFIG_DIR set, so a secondary account read the
primary's tier. The profile's tier also keeps the subscription in the label,
so a Team seat stays "Team" (#11109).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Call a lapsed Claude access token paused, not signed out (#8093)

* Refresh Claude usage after the clock moves backwards (#9956)

* Bound unreadable Claude transcript warnings (#12414)

* Count Claude usage from opencode v2 sessions (#13894)

* Reload agent usage records when an inotify watch fails to rearm (#10067)

* Reload agent usage records after each update run instead of on a timer

Rather than #10067's two-minute timer per record, reload every record when
the omarchy-agent-usage-update process exits, the moment its files can have
been replaced. A reload that finds a file unchanged keeps its record, so the
panel isn't stirred up by identical data. The grep test now runs the QML
functions.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Show the agent status when the trouble line has no help text (#8497)

* Clear stale agent login guidance after a successful probe (#8892)

* Clear the Grok login hint after a successful probe

#8892 cleared the default login hint after a successful probe in the
Claude and Codex collectors; Grok's collector had the same stale hint.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Read Fireworks credentials from pi's auth.json (#7455)

The Fireworks collector skipped pi, Omarchy's default agent, when
walking its credential ladder, so a machine signed in to Fireworks only
through pi (/login fireworks) never showed the tab. Insert the key pi
stores in $PI_CODING_AGENT_DIR/auth.json (default ~/.pi/agent) between
the firectl auth.ini and the opencode fallback.

pi keys can be literals, $ENV_VAR/${ENV_VAR} references, or !command
shell lookups. The collector resolves the first two; command lookups
stay pi-only and are skipped rather than sent to the API verbatim.

* Call a lapsed Grok access token paused, not signed out

Grok's access token lives six hours and Grok mints a new one from its
refresh token whenever it starts, so a lapsed one is routine. Reporting
it as an expired sign-in made the panel offer Sign-in required several
times a day, sending people through grok login for nothing. With a
refresh token present it now reads as paused, like Claude's.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Keep showing Grok's last limits while it sits idle

While Grok hasn't run, nothing on the machine has spent its allowance,
so with a refresh token on hand the last numbers still stand: they show
as current rather than dimmed under a status line. A weekly window that
reset in the meantime starts over at 0%, a whole number of weeks on.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Ask for a Grok sign-in once its refresh token is past 30 days

A refresh token older than Grok's 30-day sign-in can't renew anything,
so the panel offers Sign-in required again instead of showing the last
limits as current. With nothing cached yet it says to start Grok, rather
than showing an empty section without a word.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Check both ends of what the Claude index read before resuming a transcript

A transcript rewritten in place could grow and change only after its
first kilobytes, and the index took it for an append. It now compares
the last kilobytes before the resume point too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Simplify the agent usage collectors

- Codex: pass the forced-scan choice down instead of a module global, make
  the per-file reader's cache arguments required, shrink the cache record
  check, and drop guards for shapes that can't occur: an empty launcher
  path, realpath raising, mise itself being a lazy launcher, multi-line
  `mise which` output, and probing without a temp file for stderr.
- Claude: decide an append by the digest of both ends of what was read
  alone; the inode and mtime checks it made redundant are gone.
- Snapshot: the device id falls back to the hostname, which always exists.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Share fixture setup in the agent usage scanner tests

Every fixture home lives under one scratch directory with a single cleanup
trap, instead of a trap rewritten with a longer list for each new home, and
the Codex test builds its signed-in homes with one helper.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Treat a replaced Claude transcript as new even when its ends match

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* Probe Codex without its error text when there's no temporary space

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: tossbaws <17258053+tossbaws@users.noreply.github.com>
Co-authored-by: surim0n <suritech@gmail.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: anonwurcod <anonwurcod@proton.me>
Co-authored-by: Kevin Rajan <7121943+kvnloo@users.noreply.github.com>
Co-authored-by: Nate Ashby <nate.ashby11@gmail.com>
Co-authored-by: Aris Gysel <aris.gysel@me.com>
Co-authored-by: Brams <76213579+Brams-s@users.noreply.github.com>
Co-authored-by: This_Is_NPC <gabrielfollone27@gmail.com>
Co-authored-by: sanjyay <102979855+sanjyay@users.noreply.github.com>
Co-authored-by: PapistProtocol <12738904+PapistProtocol@users.noreply.github.com>
Co-authored-by: steez <stevedimakos97@gmail.com>
Co-authored-by: GPT-6 Astra <noreply@openai.com>
Co-authored-by: Ryan Yogan <ryanyogan@gmail.com>
Co-authored-by: Oli Denton <41393837+omdenton@users.noreply.github.com>
Co-authored-by: Igor Kramar <i@ikramar.ru>
Co-authored-by: Martin Eidensten <martin@meibe.se>
Co-authored-by: Romain Perron <rdj.perron@gmail.com>
Co-authored-by: Omarchy Contributor <contributor@users.noreply.github.com>
Co-authored-by: manuaudio <manu@arimaka.com>
Co-authored-by: Tyler South <tsouth2@gmail.com>
Co-authored-by: whathek <Hek846@users.noreply.github.com>
Co-authored-by: Ty Richards <me@tyrichards.com>

bin/omarchy-agent-usage-claude

@@ -122,12 +122,209 @@ def empty_bucket() -> dict[str, int]:
 # ---------------------------------------------------------------- local scan
 
 
-def scan_projects(projects_path: Path) -> dict[str, Any]:
+# Transcripts are append-only, and most of the corpus never changes between
+# ticks, so the scan keeps a per-file index: the unique usage records already
+# parsed out of each file plus the byte offset they end at. A file whose size
+# and mtime match the index is not opened; a file that grew is read from the
+# stored offset; anything else (truncated, rewritten, replaced, unknown) is
+# read from the start. Without this the whole history was re-parsed every
+# refresh, and the cost grew for the life of the machine.
+
+INDEX_VERSION = 4
+
+# How much of each end of what was read the index remembers a hash of, so a
+# file replaced or rewritten with more than it had is not mistaken for an
+# append.
+INDEX_HEAD_BYTES = 4096
+
+
+# Records hold local days, so an index is only true in the timezone that
+# wrote it; after a change every transcript is read again.
+def index_zone() -> list[Any]:
+  return [os.environ.get("TZ"), list(time.tzname), time.timezone, time.altzone]
+
+
+def load_index(index_file: Path) -> dict[str, Any]:
+  try:
+    data = json.loads(index_file.read_text(encoding="utf-8"))
+    if (
+      isinstance(data, dict)
+      and data.get("version") == INDEX_VERSION
+      and data.get("zone") == index_zone()
+      and isinstance(data.get("files"), dict)
+    ):
+      return data["files"]
+  except Exception:
+    pass
+  return {}
+
+
+# What was already read, by its first and last few kilobytes: a transcript
+# rewritten in place rather than appended to changes one or the other.
+def read_digest(path: Path, length: int) -> str:
+  digest = hashlib.sha1()
+  with path.open("rb") as handle:
+    digest.update(handle.read(min(length, INDEX_HEAD_BYTES)))
+    handle.seek(max(0, length - INDEX_HEAD_BYTES))
+    digest.update(handle.read(min(length, INDEX_HEAD_BYTES)))
+  return digest.hexdigest()
+
+
+def parse_usage_records(path: Path, entry: dict[str, Any] | None) -> dict[str, Any]:
+  """Return the index entry for one transcript, reusing `entry` where the file
+  only grew. Records are [key, model, day, input, output, cacheRead,
+  cacheWrite, session]; `day` is None when the line carries no timestamp."""
+  stat = path.stat()
+  size = stat.st_size
+  mtime = stat.st_mtime_ns
+
+  if entry and entry.get("size") == size and entry.get("mtime") == mtime:
+    return entry
+
+  # Only the same file, grown past what was read with what was read still in
+  # place, is an append. A transcript replaced by a larger one is a new file,
+  # and one rewritten in place has a different start or end.
+  if (
+    entry
+    and size > number(entry.get("size"))
+    and entry.get("file") == [stat.st_dev, stat.st_ino]
+    and entry.get("head") == read_digest(path, number(entry.get("offset")))
+  ):
+    records = list(entry.get("records") or [])
+    offset = number(entry.get("offset"))
+    line_number = number(entry.get("lines"))
+  else:
+    records = []
+    offset = 0
+    line_number = 0
+
+  # A streamed response is several lines sharing one message id, and the
+  # first line's output_tokens is a placeholder while the last carries the
+  # real count. Keep each message's line with the highest output count (the
+  # last one read wins a tie), whole: a response that fell back to another
+  # model mid-stream has lines that are separate snapshots, and mixing fields
+  # across them over-counts cache tokens and credits the wrong model.
+  positions = {record[0]: index for index, record in enumerate(records)}
+  with path.open("rb") as handle:
+    handle.seek(offset)
+    for raw in handle:
+      if not raw.endswith(b"\n"):
+        # A line still being written; pick it up whole next time.
+        break
+      offset += len(raw)
+      line_number += 1
+
+      # Cheap pre-filter before JSON parsing keeps files with unrelated
+      # lines inexpensive.
+      if b'"usage":' not in raw:
+        continue
+
+      try:
+        parsed = json.loads(raw.decode("utf-8", errors="replace"))
+      except Exception:
+        continue
+
+      message = parsed.get("message") if isinstance(parsed.get("message"), dict) else {}
+      if parsed.get("type") != "assistant" and message.get("role") != "assistant":
+        continue
+
+      usage = message.get("usage") or parsed.get("usage")
+      if not isinstance(usage, dict):
+        continue
+
+      message_id = message.get("id") or parsed.get("messageId") or ""
+      key = str(message_id) if message_id else f"{path}:{parsed.get('uuid') or parsed.get('requestId') or line_number}"
+
+      input_tokens = usage_token(usage, "input_tokens", "inputTokens")
+      output_tokens = usage_token(usage, "output_tokens", "outputTokens")
+      cache_read = usage_token(usage, "cache_read_input_tokens", "cacheReadInputTokens")
+      cache_write = usage_token(usage, "cache_creation_input_tokens", "cacheCreationInputTokens")
+      if input_tokens + output_tokens + cache_read + cache_write <= 0:
+        continue
+
+      held = positions.get(key)
+      if held is not None and output_tokens < records[held][4]:
+        continue
+
+      timestamp = parsed.get("timestamp") or message.get("timestamp")
+      record = [
+        key,
+        str(message.get("model") or parsed.get("model") or "claude"),
+        local_date_from_timestamp(timestamp) if timestamp is not None else None,
+        input_tokens,
+        output_tokens,
+        cache_read,
+        cache_write,
+        str(parsed.get("sessionId") or path),
+      ]
+      if held is None:
+        positions[key] = len(records)
+        records.append(record)
+      else:
+        records[held] = record
+
+  return {
+    "size": size,
+    "file": [stat.st_dev, stat.st_ino],
+    "mtime": mtime,
+    "head": read_digest(path, offset),
+    "offset": offset,
+    "lines": line_number,
+    "records": records,
+  }
+
+
+def scan_projects(projects_path: Path, index_file: Path | None = None) -> dict[str, Any]:
+  previous = load_index(index_file) if index_file else {}
+  current: dict[str, Any] = {}
+
+  # Directory order, as before: when one API message shows up in several
+  # transcripts with the same output count, the last file visited wins, and
+  # the numbers must not move because the scan got faster.
+  order: list[str] = []
+  unreadable_count = 0
+  unreadable_examples: list[str] = []
+  files = projects_path.rglob("*.jsonl") if projects_path.is_dir() else []
+  for path in files:
+    key = str(path)
+    try:
+      current[key] = parse_usage_records(path, previous.get(key))
+    except Exception as exc:
+      unreadable_count += 1
+      if len(unreadable_examples) < 3:
+        unreadable_examples.append(f"{path}: {exc}")
+      # What was read before is still true of the file.
+      if key in previous:
+        current[key] = previous[key]
+    if key in current:
+      order.append(key)
+
+  if unreadable_count:
+    examples = "; ".join(unreadable_examples)
+    print(
+      f"Ignoring {unreadable_count} unreadable Claude project files"
+      f" (examples: {examples})",
+      file=sys.stderr,
+    )
+
+  if index_file and current != previous:
+    try:
+      write_json(index_file, {"version": INDEX_VERSION, "zone": index_zone(), "files": current})
+    except Exception as exc:
+      print(f"Could not write Claude scan index {index_file}: {exc}", file=sys.stderr)
+
+  return summarize_records(current, order)
+
+
+def summarize_records(indexed: dict[str, Any], order: list[str]) -> dict[str, Any]:
   today = local_date_string()
   recent_dates = recent_date_strings()
   recent = {day: {"date": day, "messageCount": 0} for day in recent_dates}
 
-  seen: set[str] = set()
+  # One API message can show up in several transcripts (a resumed session
+  # carries the earlier ones). As within a file, the record with the highest
+  # output count wins, the last one visited breaking a tie.
+  seen: dict[str, list[Any]] = {}
   sessions: set[str] = set()
   active_days: set[str] = set()
   today_sessions: set[str] = set()
@@ -137,68 +334,38 @@ def scan_projects(projects_path: Path) -> dict[str, Any]:
   today_prompt_count = 0
   today_token_total = 0
 
-  files = projects_path.rglob("*.jsonl") if projects_path.is_dir() else []
-  for path in files:
-    try:
-      with path.open("r", encoding="utf-8", errors="replace") as handle:
-        for line_number, line in enumerate(handle, 1):
-          # Cheap pre-filter before JSON parsing keeps files with unrelated
-          # lines inexpensive.
-          if '"usage":' not in line:
-            continue
+  for key in order:
+    for record in indexed[key].get("records") or []:
+      held = seen.get(record[0])
+      if held is not None and record[4] < held[4]:
+        continue
+      seen[record[0]] = record
 
-          try:
-            entry = json.loads(line)
-          except Exception:
-            continue
+  for _, model, day, input_tokens, output_tokens, cache_read, cache_write, session_key in seen.values():
+    if day is None:
+      day = today
+    total = input_tokens + output_tokens + cache_read + cache_write
 
-          message = entry.get("message") if isinstance(entry.get("message"), dict) else {}
-          if entry.get("type") != "assistant" and message.get("role") != "assistant":
-            continue
+    sessions.add(session_key)
+    active_days.add(day)
+    prompts += 1
 
-          usage = message.get("usage") or entry.get("usage")
-          if not isinstance(usage, dict):
-            continue
+    bucket = usage_by_model.setdefault(model, empty_bucket())
+    bucket["inputTokens"] += input_tokens
+    bucket["outputTokens"] += output_tokens
+    bucket["cacheReadInputTokens"] += cache_read
+    bucket["cacheCreationInputTokens"] += cache_write
 
-          message_id = message.get("id") or entry.get("messageId") or ""
-          unique_key = str(message_id) if message_id else f"{path}:{entry.get('uuid') or entry.get('requestId') or line_number}"
-          if unique_key in seen:
-            continue
-          seen.add(unique_key)
+    if day in recent:
+      # recentDays.messageCount is actually a token total, despite the
+      # legacy name shared with synced snapshots.
+      recent[day]["messageCount"] += total
 
-          input_tokens = usage_token(usage, "input_tokens", "inputTokens")
-          output_tokens = usage_token(usage, "output_tokens", "outputTokens")
-          cache_read = usage_token(usage, "cache_read_input_tokens", "cacheReadInputTokens")
-          cache_write = usage_token(usage, "cache_creation_input_tokens", "cacheCreationInputTokens")
-          total = input_tokens + output_tokens + cache_read + cache_write
-          if total <= 0:
-            continue
-
-          model = str(message.get("model") or entry.get("model") or "claude")
-          day = local_date_from_timestamp(entry.get("timestamp") or message.get("timestamp"))
-          session_key = str(entry.get("sessionId") or path)
-          sessions.add(session_key)
-          active_days.add(day)
-          prompts += 1
-
-          bucket = usage_by_model.setdefault(model, empty_bucket())
-          bucket["inputTokens"] += input_tokens
-          bucket["outputTokens"] += output_tokens
-          bucket["cacheReadInputTokens"] += cache_read
-          bucket["cacheCreationInputTokens"] += cache_write
-
-          if day in recent:
-            # recentDays.messageCount is actually a token total, despite the
-            # legacy name shared with synced snapshots.
-            recent[day]["messageCount"] += total
-
-          if day == today:
-            today_prompt_count += 1
-            today_sessions.add(session_key)
-            today_token_total += total
-            today_tokens[model] = today_tokens.get(model, 0) + total
-    except Exception as exc:
-      print(f"Ignoring unreadable Claude project file {path}: {exc}", file=sys.stderr)
+    if day == today:
+      today_prompt_count += 1
+      today_sessions.add(session_key)
+      today_token_total += total
+      today_tokens[model] = today_tokens.get(model, 0) + total
 
   return {
     "todayPrompts": today_prompt_count,
@@ -217,18 +384,28 @@ def scan_projects(projects_path: Path) -> dict[str, Any]:
   }
 
 
-def scan_cache_paths(projects_path: Path) -> tuple[Path, Path]:
+def scan_cache_paths(projects_path: Path) -> tuple[Path, Path, Path]:
   digest = hashlib.sha1(str(projects_path).encode("utf-8")).hexdigest()[:16]
   root = cache_root()
-  return root / f"claude-scan-{digest}.json", root / f"claude-scan-{digest}.lock"
+  return root / f"claude-scan-{digest}.json", root / f"claude-scan-{digest}.lock", root / f"claude-index-{digest}.json"
+
+
+def read_json(path: Path) -> dict[str, Any] | None:
+  try:
+    return json.loads(path.read_text(encoding="utf-8"))
+  except Exception:
+    return None
 
 
 def read_fresh_json(path: Path, max_age_seconds: float) -> dict[str, Any] | None:
   if max_age_seconds <= 0 or not path.exists():
     return None
   try:
-    if time.time() - path.stat().st_mtime <= max_age_seconds:
-      return json.loads(path.read_text(encoding="utf-8"))
+    # A negative age means the mtime is in the future: the clock moved
+    # backwards since the write, so the cache's freshness cannot be trusted.
+    age = time.time() - path.stat().st_mtime
+    if 0 <= age <= max_age_seconds:
+      return read_json(path)
   except Exception:
     return None
   return None
@@ -252,8 +429,8 @@ def write_json(path: Path, payload: dict[str, Any]) -> None:
     raise
 
 
-def cached_scan(projects_path: Path, max_age_seconds: float) -> dict[str, Any]:
-  cache_file, lock_file = scan_cache_paths(projects_path)
+def cached_scan(projects_path: Path, max_age_seconds: float, use_index: bool = True) -> dict[str, Any]:
+  cache_file, lock_file, index_file = scan_cache_paths(projects_path)
 
   cached = read_fresh_json(cache_file, max_age_seconds)
   if cached is not None:
@@ -264,7 +441,9 @@ def cached_scan(projects_path: Path, max_age_seconds: float) -> dict[str, Any]:
     cached = read_fresh_json(cache_file, max_age_seconds)
     if cached is not None:
       return cached
-    summary = scan_projects(projects_path)
+    if not use_index:
+      index_file.unlink(missing_ok=True)
+    summary = scan_projects(projects_path, index_file)
     write_json(cache_file, summary)
     return summary
 
@@ -341,11 +520,27 @@ def today_prompts_from_history(claude_dir: Path) -> tuple[int, int]:
 # provider, model, and token usage on every assistant message.
 
 
+# `omp --profile=<name>` (and pi's equivalent) relocates the whole agent tree
+# under <base>/profiles/<name>/, so a subscription driven entirely through a
+# profile leaves the default root empty and its usage uncounted. Sessions are
+# keyed by file path, so a profile adds sessions rather than double-counting
+# the default root.
+def pi_session_roots() -> list[Path]:
+  roots: list[Path] = []
+  for base in (Path.home() / ".pi", Path.home() / ".omp"):
+    roots.append(base / "agent" / "sessions")
+    profiles = base / "profiles"
+    try:
+      # Sorted so the scan order does not depend on directory order.
+      roots.extend(sorted(child / "agent" / "sessions" for child in profiles.iterdir() if child.is_dir()))
+    except OSError:
+      # No profiles directory, or it is unreadable: the default root stands.
+      pass
+  return roots
+
+
 def scan_pi_usage(max_age_seconds: float) -> dict[str, Any] | None:
-  roots = [
-    Path.home() / ".pi" / "agent" / "sessions",
-    Path.home() / ".omp" / "agent" / "sessions",
-  ]
+  roots = pi_session_roots()
   cache_file = cache_root() / "claude-pi-sessions.json"
   cached = read_fresh_json(cache_file, max_age_seconds)
   if cached is not None:
@@ -477,16 +672,36 @@ def scan_opencode_usage(max_age_seconds: float) -> dict[str, Any] | None:
     return None
   try:
     conn.execute("PRAGMA query_only = ON")
-    for session_id, raw in conn.execute("SELECT session_id, data FROM message"):
+    tables = {name for (name,) in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'")}
+    # opencode v1 keeps messages in `message`, with the role, provider and model
+    # in each row's data. v2 writes `session_message` instead: the role moves to
+    # the `type` column and the provider and model nest under data.model. A
+    # database can hold either or both.
+    queries = []
+    if "message" in tables:
+      queries.append(("opencode:", "SELECT session_id, data, NULL FROM message"))
+    if "session_message" in tables:
+      queries.append(("opencode-v2:", "SELECT session_id, data, type FROM session_message"))
+    rows = ((prefix, row) for prefix, query in queries for row in conn.execute(query))
+    for session_prefix, (session_id, raw, row_type) in rows:
       # One malformed row must not abort the scan, so every shape assumption
       # lives inside the try.
       try:
         entry = json.loads(raw)
+        if not isinstance(entry, dict):
+          continue
+        if row_type is None:
+          role = entry.get("role")
+          provider = entry.get("providerID")
+          model_id = entry.get("modelID")
+        else:
+          role = row_type
+          model_info = entry.get("model") or {}
+          provider = model_info.get("providerID")
+          model_id = model_info.get("id")
         # Exact match: opencode provider ids are free-form, and a custom
         # "anthropic-proxy" gateway is not this subscription.
-        if not isinstance(entry, dict) or entry.get("role") != "assistant":
-          continue
-        if str(entry.get("providerID") or "") != "anthropic":
+        if role != "assistant" or str(provider or "") != "anthropic":
           continue
         tokens = entry.get("tokens") or {}
         cache = tokens.get("cache") or {}
@@ -501,10 +716,10 @@ def scan_opencode_usage(max_age_seconds: float) -> dict[str, Any] | None:
 
         created = number((entry.get("time") or {}).get("created"))
         day = dt.datetime.fromtimestamp(created / 1000).strftime("%Y-%m-%d") if created > 0 else today
-        model = str(entry.get("modelID") or "claude").rstrip("/").split("/")[-1]
+        model = str(model_id or "claude").rstrip("/").split("/")[-1]
       except Exception:
         continue
-      session_key = "opencode:" + str(session_id)
+      session_key = session_prefix + str(session_id)
       sessions.add(session_key)
       active_days.add(day)
       prompts += 1
@@ -585,28 +800,75 @@ def merge_stats(base: dict[str, Any], extra: dict[str, Any]) -> dict[str, Any]:
 # CLI's login. Nothing else leaves the credential store: the token goes
 # nowhere but the Authorization header of the limits probe, and only the
 # plan label may travel into the printed record.
-def oauth_login(claude_dir: Path) -> tuple[str, int, str]:
+def oauth_login(claude_dir: Path) -> tuple[str, int, int, str]:
   try:
     data = json.loads((claude_dir / ".credentials.json").read_text(encoding="utf-8"))
   except Exception:
-    return "", 0, ""
+    return "", 0, 0, ""
   login = data.get("claudeAiOauth")
   if not isinstance(login, dict):
-    return "", 0, ""
-  plan = plan_label(str(login.get("rateLimitTier") or ""), str(login.get("subscriptionType") or ""))
-  return str(login.get("accessToken") or ""), number(login.get("expiresAt")), plan
+    return "", 0, 0, ""
+  # The profile's tier is asked first, but only one that names a multiplier
+  # outranks the credentials'. The subscription still leads the label, so a
+  # Team seat keeps reading Team.
+  tier = profile_tier(claude_dir)
+  if not re.search(r"max_(\d+x)", tier, re.IGNORECASE):
+    tier = str(login.get("rateLimitTier") or "")
+  plan = plan_label(tier, str(login.get("subscriptionType") or ""))
+  return (
+    str(login.get("accessToken") or ""),
+    number(login.get("expiresAt")),
+    number(login.get("refreshTokenExpiresAt")),
+    plan,
+  )
 
 
-def plan_label(tier: str, subscription: str) -> str:
-  if tier:
-    match = re.search(r"max_(\d+x)", tier, re.IGNORECASE)
-    if match:
-      return "Max " + match.group(1)
-  if subscription:
-    return subscription[0].upper() + subscription[1:]
+# The credentials carry the tier their token was minted with, and an upgrade
+# does not restamp it: a refresh rewrites the file and keeps the old value, so
+# an account moved to Max 20x keeps reading Max 5x. The CLI's own profile is
+# refreshed against the account, so the tier it states there is asked first.
+def profile_tier(claude_dir: Path) -> str:
+  try:
+    account = json.loads(profile_path(claude_dir).read_text(encoding="utf-8")).get("oauthAccount")
+  except Exception:
+    return ""
+  if not isinstance(account, dict):
+    return ""
+  # A seat with no tier of its own is limited by its organization, which is
+  # the shape a personal Max subscription takes.
+  for key in ("userRateLimitTier", "organizationRateLimitTier"):
+    tier = str(account.get(key) or "")
+    if tier:
+      return tier
   return ""
 
 
+# The CLI's profile for a config home: a legacy .config.json inside it, else
+# ~/.claude.json for the default home and <home>/.claude.json for one it was
+# pointed at with CLAUDE_CONFIG_DIR. A copy anywhere else is never refreshed,
+# and another home's profile describes another account.
+def profile_path(home: Path) -> Path:
+  legacy = home / ".config.json"
+  if legacy.exists():
+    return legacy
+  configured = os.environ.get("CLAUDE_CONFIG_DIR")
+  if home == expand_path("~/.claude") and not (configured and expand_path(configured) == home):
+    return Path.home() / ".claude.json"
+  return home / ".claude.json"
+
+
+# The rate-limit tier names the ceiling, not the plan: a Team premium seat
+# runs on `default_claude_max_5x` while paying for Team. Lead with the
+# subscription and keep the multiplier as its qualifier, so Max still reads
+# "Max 5x" and a Team seat reads "Team 5x" rather than passing as Max.
+def plan_label(tier: str, subscription: str) -> str:
+  plan = subscription[0].upper() + subscription[1:] if subscription else ""
+  match = re.search(r"max_(\d+x)", tier, re.IGNORECASE) if tier else None
+  if not match:
+    return plan
+  return (plan or "Max") + " " + match.group(1)
+
+
 def parse_utilization(value: Any) -> float:
   try:
     return float(str(value).strip().replace("%", ""))
@@ -795,7 +1057,7 @@ def usable_cached_limits(cached: dict[str, Any]) -> list[dict[str, Any]]:
   return [entry for entry in entries if isinstance(entry, dict) and limit_window_open(entry, now)]
 
 
-def collect_limits(access_token: str, expires_at_ms: int, force: bool, cache_name: str = "claude-limits.json") -> dict[str, Any]:
+def collect_limits(access_token: str, expires_at_ms: int, refresh_expires_at_ms: int, force: bool, cache_name: str = "claude-limits.json") -> dict[str, Any]:
   # `live` says the numbers came from Anthropic just now (or within the reuse
   # window) rather than from a cache kept past a failed or impossible probe.
   result = {"limits": [], "usageStatusText": "", "authHelpText": AUTH_HELP, "live": False}
@@ -809,7 +1071,7 @@ def collect_limits(access_token: str, expires_at_ms: int, force: bool, cache_nam
   # kept as the answer of record when a later probe fails. Every account
   # keeps its own, so one account's allowance never stands in for another's.
   probe_cache = cache_root() / cache_name
-  cached = read_fresh_json(probe_cache, float("inf")) or {}
+  cached = read_json(probe_cache) or {}
   fallback = usable_cached_limits(cached)
   result["fetchedAtMs"] = number(cached.get("fetchedAtMs"))
 
@@ -824,6 +1086,19 @@ def collect_limits(access_token: str, expires_at_ms: int, force: bool, cache_nam
     return result
   if expires_at_ms > 0 and expires_at_ms <= time.time() * 1000:
     result["limits"] = fallback
+    # A lapsed access token is routine, not a sign-out: it lives about eight
+    # hours, and the CLI mints a new one from the refresh token the next time
+    # it runs. Only a lapsed *refresh* token is a sign-in a person must redo,
+    # and saying so either way sends people to `claude auth login` nightly --
+    # which does not restore anything the CLI would not have restored itself.
+    if refresh_expires_at_ms > time.time() * 1000:
+      result["usageStatusText"] = "Limits paused"
+      result["authHelpText"] = (
+        "Claude Code's access token has lapsed"
+        + (" — showing the last known limits." if fallback else ".")
+        + " Start Claude Code to refresh it; signing in again is not needed."
+      )
+      return result
     result["usageStatusText"] = "Sign-in expired"
     result["authHelpText"] = (
       "Claude Code's saved sign-in expired"
@@ -836,15 +1111,18 @@ def collect_limits(access_token: str, expires_at_ms: int, force: bool, cache_nam
   # entirely; the interval is there to absorb repeated panel opens, not to
   # overrule someone who pressed refresh.
   fetched_at = number(cached.get("fetchedAtMs")) / 1000
-  if fallback and not force and time.time() - fetched_at < PROBE_MIN_INTERVAL_SECONDS:
+  fetched_age = time.time() - fetched_at
+  if fallback and not force and 0 <= fetched_age < PROBE_MIN_INTERVAL_SECONDS:
     result["limits"] = fallback
     result["live"] = True
+    result["authHelpText"] = ""
     return result
 
   probe = probe_limits(access_token)
   if probe["ok"]:
     result["limits"] = probe["limits"]
     result["live"] = True
+    result["authHelpText"] = ""
     result["fetchedAtMs"] = round(time.time() * 1000)
     write_json(probe_cache, {"fetchedAtMs": result["fetchedAtMs"], "limits": probe["limits"]})
     return result
@@ -855,7 +1133,8 @@ def collect_limits(access_token: str, expires_at_ms: int, force: bool, cache_nam
     result["retryAdvised"] = True
   if fallback:
     result["limits"] = fallback
-    result["live"] = time.time() - result["fetchedAtMs"] / 1000 < CURRENT_ENOUGH_SECONDS
+    fetched_age = time.time() - result["fetchedAtMs"] / 1000
+    result["live"] = 0 <= fetched_age < CURRENT_ENOUGH_SECONDS
   else:
     result["usageStatusText"] = "Claude limits unavailable"
     result["authHelpText"] = probe["helpText"]
@@ -895,7 +1174,7 @@ def registered_accounts() -> list[dict[str, Any]]:
 # know this account is the one with the most room. Each window it last saw
 # comes back at 0%.
 def replenished_limits(cache_name: str) -> list[dict[str, Any]]:
-  cached = read_fresh_json(cache_root() / cache_name, float("inf")) or {}
+  cached = read_json(cache_root() / cache_name) or {}
   entries = cached.get("limits")
   if not isinstance(entries, list):
     return []
@@ -908,9 +1187,8 @@ def replenished_limits(cache_name: str) -> list[dict[str, Any]]:
 
 
 def current_account_id(home: Path) -> str:
-  path = Path.home() / ".claude.json" if home == expand_path("~/.claude") else home / ".claude.json"
   try:
-    account = json.loads(path.read_text(encoding="utf-8")).get("oauthAccount") or {}
+    account = json.loads(profile_path(home).read_text(encoding="utf-8")).get("oauthAccount") or {}
   except Exception:
     return ""
   return str(account.get("accountUuid") or "")
@@ -918,7 +1196,7 @@ def current_account_id(home: Path) -> str:
 
 def account_limits(account: dict[str, Any], force: bool) -> dict[str, Any]:
   home = expand_path(account["home"]) if account.get("home") else expand_path("~/.claude")
-  access_token, expires_at_ms, plan = oauth_login(home)
+  access_token, expires_at_ms, refresh_expires_at_ms, plan = oauth_login(home)
   # Keyed by the subscription, not the label: forget "Work" and add a different
   # "Work", and the new one must never fall back on the old one's allowance.
   # Who the home is signed in as now, not when it was added: a `claude auth
@@ -932,7 +1210,7 @@ def account_limits(account: dict[str, Any], force: bool) -> dict[str, Any]:
     cache_name = "claude-limits.json"
   else:
     cache_name = f"claude-limits-{key}.json"
-  limits = collect_limits(access_token, expires_at_ms, force, cache_name)
+  limits = collect_limits(access_token, expires_at_ms, refresh_expires_at_ms, force, cache_name)
   if not limits["live"] and not limits["limits"]:
     limits["limits"] = replenished_limits(cache_name)
   return {
@@ -956,14 +1234,14 @@ def account_limits(account: dict[str, Any], force: bool) -> dict[str, Any]:
 
 def main() -> int:
   parser = argparse.ArgumentParser()
-  parser.add_argument("--force", action="store_true", help="rescan transcripts and re-probe limits, ignoring caches")
+  parser.add_argument("--force", action="store_true", help="rescan transcripts from scratch and re-probe limits, ignoring caches")
   parser.add_argument("--limits-only", action="store_true", help="reuse any recent transcript scan; only the limits probe must be fresh")
   parser.add_argument("--cache-seconds", type=float, default=20)
   args = parser.parse_args()
 
   claude_dir = config_dir()
   scan_age = 0 if args.force else (900 if args.limits_only else args.cache_seconds)
-  stats = cached_scan(claude_dir / "projects", scan_age)
+  stats = cached_scan(claude_dir / "projects", scan_age, use_index=not args.force)
 
   if number(stats.get("totalPrompts")) <= 0:
     fallback = stats_cache_fallback(claude_dir)
@@ -994,8 +1272,8 @@ def main() -> int:
     limits = {key: current[key] for key in ("limits", "usageStatusText", "authHelpText", "retryAdvised")}
     stale, fetched_at = current["stale"], current["fetchedAt"]
   else:
-    access_token, expires_at_ms, plan = oauth_login(claude_dir)
-    limits = collect_limits(access_token, expires_at_ms, args.force)
+    access_token, expires_at_ms, refresh_expires_at_ms, plan = oauth_login(claude_dir)
+    limits = collect_limits(access_token, expires_at_ms, refresh_expires_at_ms, args.force)
     stale, fetched_at = not limits["live"], limits["fetchedAtMs"]
 
   record = {

bin/omarchy-agent-usage-codex

@@ -22,6 +22,7 @@ import subprocess
 import sys
 import tempfile
 import time
+import tomllib
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
@@ -29,6 +30,11 @@ AGENT_ID = "codex"
 AGENT_NAME = "Codex"
 AUTH_HELP = "Run `codex login` to authenticate."
 
+# session_meta.model_provider for the built-in OpenAI backend, which is the
+# only one this subscription pays for. Any other id points somewhere else:
+# `--oss`'s ollama or lmstudio, or a custom provider in config.toml.
+NATIVE_PROVIDER = "openai"
+
 # A scan this recent is only reused to dedup concurrent collector runs (the
 # update command backgrounds one per agent while the panel refreshes on its
 # own); every periodic widget refresh lands a real rescan, however low
@@ -37,6 +43,11 @@ AUTH_HELP = "Run `codex login` to authenticate."
 SCAN_REUSE_SECONDS = 20
 LIMITS_ONLY_REUSE_SECONDS = 900
 
+# Per-file totals of native Codex sessions survive between runs, so a rescan
+# only reads the session files that changed. Bumping the schema retires every
+# record written by an older layout.
+FILE_CACHE_SCHEMA = 1
+
 
 def local_day(value):
   if value is None:
@@ -91,6 +102,52 @@ def find_command(name):
   return shutil.which(name, path=ENV.get("PATH"))
 
 
+def mise_shims_dir():
+  data_home = ENV.get("MISE_DATA_DIR") or os.path.join(
+    ENV.get("XDG_DATA_HOME") or os.path.join(str(Path.home()), ".local", "share"), "mise")
+  return os.path.join(data_home, "shims")
+
+
+def is_lazy_launcher(path):
+  # omarchy-mise-install writes a regular file whose first step is `mise use
+  # -g`, and mise's shims exec `mise x`; running either installs the tool it
+  # wraps. A shim is a symlink to the mise binary itself, so the symlink
+  # check comes after the shim checks: a symlink elsewhere is the user's own
+  # binary.
+  if os.path.dirname(os.path.abspath(path)) == mise_shims_dir():
+    return True
+  if os.path.islink(path):
+    mise = find_command("mise")
+    return bool(mise) and os.path.realpath(path) == os.path.realpath(mise)
+  try:
+    with open(path, "rb") as handle:
+      head = handle.read(4096)
+    return b"mise use" in head or b"mise x " in head
+  except OSError:
+    return False
+
+
+def find_codex_binary():
+  for directory in ENV.get("PATH", "").split(os.pathsep):
+    candidate = os.path.join(directory, "codex")
+    if os.path.isfile(candidate) and os.access(candidate, os.X_OK) and not is_lazy_launcher(candidate):
+      return candidate
+  # `mise which` only prints a binary that is already installed, so it is the
+  # safe way to resolve a codex managed by mise without running the launcher.
+  # `mise which` can resolve a configured `latest` over the network; the usage
+  # probe stays offline.
+  mise = find_command("mise")
+  if mise:
+    try:
+      out = subprocess.run([mise, "which", "codex"], capture_output=True, text=True, env=dict(ENV, MISE_OFFLINE="1"), timeout=10)
+      resolved = out.stdout.strip()
+      if out.returncode == 0 and os.path.isfile(resolved) and not is_lazy_launcher(resolved):
+        return resolved
+    except Exception:
+      pass
+  return None
+
+
 now = datetime.now()
 today = now.strftime("%Y-%m-%d")
 recent_dates = [(now - timedelta(days=offset)).strftime("%Y-%m-%d") for offset in range(6, -1, -1)]
@@ -107,46 +164,109 @@ total_sessions = set()
 seen_pi_messages = set()
 
 
-def add_usage(day, session_key, model, input_tokens, output_tokens, cache_read, cache_write):
+# One session file's totals, the unit the file cache stores: day -> model ->
+# [input, output, cacheRead, cacheWrite, prompts]. Day-keyed and with no
+# "today" in it, so a record stays true however long it is kept.
+def new_file_record(stat_result):
+  return {"mtime": stat_result.st_mtime, "size": stat_result.st_size, "days": {}}
+
+
+def record_usage(record, day, model, input_tokens, output_tokens, cache_read, cache_write):
+  totals = record["days"].setdefault(day, {}).setdefault(model, [0, 0, 0, 0, 0])
+  totals[0] += input_tokens
+  totals[1] += output_tokens
+  totals[2] += cache_read
+  totals[3] += cache_write
+  totals[4] += 1
+
+
+def merge_file_record(session_key, record):
+  """Fold one file's totals, freshly read or replayed from cache, into the run."""
   global today_prompts, today_total_tokens, total_prompts
-  total = input_tokens + output_tokens + cache_read + cache_write
-  total_prompts += 1
-  total_sessions.add(session_key)
-  active_days.add(day)
+  for day, models in record["days"].items():
+    for model, totals in models.items():
+      input_tokens, output_tokens, cache_read, cache_write, prompts = totals
+      total = input_tokens + output_tokens + cache_read + cache_write
+      total_prompts += prompts
+      total_sessions.add(session_key)
+      active_days.add(day)
 
-  bucket = model_usage.setdefault(model, {
-    "inputTokens": 0,
-    "outputTokens": 0,
-    "cacheReadInputTokens": 0,
-    "cacheCreationInputTokens": 0,
-  })
-  bucket["inputTokens"] += input_tokens
-  bucket["outputTokens"] += output_tokens
-  bucket["cacheReadInputTokens"] += cache_read
-  bucket["cacheCreationInputTokens"] += cache_write
+      bucket = model_usage.setdefault(model, {
+        "inputTokens": 0,
+        "outputTokens": 0,
+        "cacheReadInputTokens": 0,
+        "cacheCreationInputTokens": 0,
+      })
+      bucket["inputTokens"] += input_tokens
+      bucket["outputTokens"] += output_tokens
+      bucket["cacheReadInputTokens"] += cache_read
+      bucket["cacheCreationInputTokens"] += cache_write
 
-  if day in recent:
-    recent[day]["messageCount"] += total
+      if day in recent:
+        recent[day]["messageCount"] += total
 
-  if day == today:
-    today_prompts += 1
-    today_sessions.add(session_key)
-    today_total_tokens += total
-    today_tokens_by_model[model] = today_tokens_by_model.get(model, 0) + total
+      if day == today:
+        today_prompts += prompts
+        today_sessions.add(session_key)
+        today_total_tokens += total
+        today_tokens_by_model[model] = today_tokens_by_model.get(model, 0) + total
+
+
+def add_usage(day, session_key, model, input_tokens, output_tokens, cache_read, cache_write):
+  record = {"days": {}}
+  record_usage(record, day, model, input_tokens, output_tokens, cache_read, cache_write)
+  merge_file_record(session_key, record)
+
+
+def pi_parent_session(path):
+  try:
+    with open(path, encoding="utf-8", errors="replace") as session_file:
+      header = json.loads(session_file.readline())
+  except (OSError, json.JSONDecodeError):
+    return None
+
+  if not isinstance(header, dict):
+    return None
+  parent = header.get("parentSession")
+  if not isinstance(parent, str) or not parent:
+    return None
+  return os.path.abspath(os.path.expanduser(parent))
+
+
+# `omp --profile=<name>` (and pi's equivalent) relocates the whole agent tree
+# under <base>/profiles/<name>/, so a subscription driven entirely through a
+# profile leaves the default root empty and its usage uncounted. Each root is
+# scanned separately and sessions are keyed by path, so a profile adds
+# sessions rather than double-counting the default one.
+def pi_session_roots():
+  roots = []
+  for base in (Path.home() / ".pi", Path.home() / ".omp"):
+    roots.append(base / "agent" / "sessions")
+    profiles = base / "profiles"
+    try:
+      # Sorted so the scan order does not depend on directory order.
+      roots.extend(sorted(child / "agent" / "sessions" for child in profiles.iterdir() if child.is_dir()))
+    except OSError:
+      # No profiles directory, or it is unreadable: the default root stands.
+      pass
+  return roots
 
 
 def scan_pi_sessions():
-  roots = [
-    Path.home() / ".pi" / "agent" / "sessions",
-    Path.home() / ".omp" / "agent" / "sessions",
-  ]
+  roots = pi_session_roots()
   rg = find_command("rg") or "rg"
+  session_keys = {}
+  today_session_keys = {}
   for root in roots:
     if not root.exists():
       continue
     try:
       proc = subprocess.Popen(
-        [rg, "--json", "-e", r'"provider"\s*:\s*"openai-codex"', "-e", r'"api"\s*:\s*"openai-codex', str(root)],
+        # --no-ignore: these are data files, not a source tree. When $HOME is
+        # itself a git checkout (a common dotfiles setup), the parent repo's
+        # ignore rules would otherwise silently exclude every session file and
+        # the scan would count zero usage.
+        [rg, "--json", "--no-ignore", "-e", r'"provider"\s*:\s*"openai-codex"', "-e", r'"api"\s*:\s*"openai-codex', str(root)],
         stdout=subprocess.PIPE,
         stderr=subprocess.DEVNULL,
         text=True,
@@ -163,16 +283,22 @@ def scan_pi_sessions():
         if event.get("type") != "match":
           continue
         line = event.get("data", {}).get("lines", {}).get("text", "")
-        path = event.get("data", {}).get("path", {}).get("text", "pi-session")
+        raw_path = event.get("data", {}).get("path", {}).get("text", "pi-session")
+        path = os.path.abspath(os.path.expanduser(raw_path))
         entry = json.loads(line)
       except Exception:
         continue
 
       if entry.get("type") != "message":
         continue
-      message_key = path + ":" + str(entry.get("id") or "")
-      if message_key in seen_pi_messages:
-        continue
+      message_id = str(entry.get("id") or "")
+      message_timestamp = str(entry.get("timestamp") or "")
+      # Forks retain both fields; IDs alone can collide across sessions.
+      if message_id and message_timestamp:
+        message_key = ("entry", message_id, message_timestamp)
+      else:
+        message_key = ("path", path, message_id)
+      duplicate = message_key in seen_pi_messages
       seen_pi_messages.add(message_key)
       message = entry.get("message") or {}
       if message.get("role") != "assistant":
@@ -196,14 +322,37 @@ def scan_pi_sessions():
         continue
 
       day = local_day(entry.get("timestamp") or message.get("timestamp"))
-      session_key = path
-      add_usage(day, session_key, model_name(message.get("model")), input_tokens, output_tokens, cache_read, cache_write)
+      session_keys.setdefault(path, set()).add(message_key)
+      if day == today:
+        today_session_keys.setdefault(path, set()).add(message_key)
+      if duplicate:
+        continue
+      add_usage(day, path, model_name(message.get("model")), input_tokens, output_tokens, cache_read, cache_write)
 
     try:
       proc.wait(timeout=1)
     except Exception:
       proc.kill()
 
+  # Resolve session ownership after scanning so rg's file order cannot affect it.
+  parents = {path: pi_parent_session(path) for path in session_keys}
+  total_sessions.difference_update(session_keys)
+  today_sessions.difference_update(session_keys)
+  for path, keys in session_keys.items():
+    inherited = set()
+    visited = {path}
+    parent = parents.get(path)
+    while parent and parent not in visited:
+      visited.add(parent)
+      inherited.update(session_keys.get(parent, ()))
+      parent = parents.get(parent)
+
+    unique_keys = keys - inherited
+    if unique_keys:
+      total_sessions.add(path)
+    if unique_keys & today_session_keys.get(path, set()):
+      today_sessions.add(path)
+
 
 def scan_opencode_sessions():
   # A subscription burned entirely through opencode leaves no native session
@@ -283,7 +432,7 @@ def scan_opencode_sessions():
   return True
 
 
-def scan_native_codex_sessions():
+def scan_native_codex_sessions(cached_files, scanned_files):
   codex_home = Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex"))
   roots = [codex_home / "sessions", codex_home / "archived_sessions"]
   files = []
@@ -293,48 +442,98 @@ def scan_native_codex_sessions():
       continue
     for path in root.rglob("*.jsonl"):
       try:
-        if path.stat().st_mtime >= cutoff:
-          files.append(path)
+        stat_result = path.stat()
       except OSError:
-        pass
+        continue
+      if stat_result.st_mtime >= cutoff:
+        files.append((path, stat_result))
 
-  for path in files:
-    current_model = "codex"
-    try:
-      with path.open(errors="replace") as handle:
-        for raw in handle:
-          try:
-            entry = json.loads(raw)
-          except Exception:
+  for path, stat_result in files:
+    key = str(path)
+    record = reusable_record(cached_files, key, stat_result)
+    if record is None:
+      record = read_native_codex_session(path, stat_result)
+    if record["mtime"] is not None:
+      scanned_files[key] = record
+    merge_file_record(key, record)
+
+
+def read_native_codex_session(path, stat_result):
+  record = new_file_record(stat_result)
+  current_model = "codex"
+  seen_meta = False
+  previous_total_usage = None
+  try:
+    with path.open(errors="replace") as handle:
+      for raw in handle:
+        # Cheap pre-filter before JSON parsing keeps files with unrelated
+        # lines inexpensive. session_meta still has to get through: it says
+        # which provider served the rollout.
+        if '"token_count"' not in raw and '"turn_context"' not in raw and '"session_meta"' not in raw:
+          continue
+        try:
+          entry = json.loads(raw)
+        except Exception:
+          continue
+        if entry.get("type") == "session_meta":
+          # A forked rollout copies its parent's session_meta after its own,
+          # so only the first one, as Codex reads it, says who served this.
+          if seen_meta:
             continue
-          if entry.get("type") == "turn_context":
-            payload = entry.get("payload") or {}
-            current_model = model_name(payload.get("model") or payload.get("model_slug") or current_model)
+          seen_meta = True
+          # Codex CLI happily fronts any OpenAI-compatible backend (`--oss`,
+          # or a custom model_provider in config.toml). Those turns bill the
+          # local box or a third party, never this subscription, so drop the
+          # whole rollout the way the pi and opencode scans drop foreign
+          # providers.
+          meta = entry.get("payload") or {}
+          provider = str(meta.get("model_provider") or "")
+          # Rollouts predating the field carry no provider at all; count
+          # those rather than silently lose the history.
+          if provider and provider != NATIVE_PROVIDER:
+            break
+          continue
+        if entry.get("type") == "turn_context":
+          payload = entry.get("payload") or {}
+          current_model = model_name(payload.get("model") or payload.get("model_slug") or current_model)
+          continue
+        payload = entry.get("payload") or entry
+        if entry.get("type") == "response_item" and isinstance(payload, dict):
+          payload = payload.get("payload") or payload
+        if not isinstance(payload, dict):
+          continue
+        if payload.get("type") != "token_count":
+          continue
+        info = payload.get("info") or {}
+        # total_token_usage is cumulative for the session. Adding every
+        # snapshot makes usage grow quadratically, so count the last turn.
+        usage = info.get("last_token_usage") or {}
+        cache_read = number(usage.get("cached_input_tokens"))
+        cache_write = number(usage.get("cache_write_input_tokens"))
+        # Cached tokens are included in input_tokens, and reasoning tokens
+        # are included in output_tokens. Keep the cache split without
+        # counting either category twice.
+        input_tokens = max(0, number(usage.get("input_tokens")) - cache_read - cache_write)
+        output_tokens = number(usage.get("output_tokens"))
+        if not (input_tokens or output_tokens or cache_read or cache_write):
+          continue
+        # Quota updates can repeat the previous request's last_token_usage.
+        # Only an unchanged cumulative snapshot proves this is a repeat:
+        # separate requests may have identical last-token counts, and older
+        # records may not carry cumulative counters at all. Keep this state
+        # per rollout and accept changed counters, including counter resets.
+        total_usage = info.get("total_token_usage")
+        if isinstance(total_usage, dict) and total_usage:
+          if total_usage == previous_total_usage:
             continue
-          payload = entry.get("payload") or entry
-          if entry.get("type") == "response_item" and isinstance(payload, dict):
-            payload = payload.get("payload") or payload
-          if not isinstance(payload, dict):
-            continue
-          if payload.get("type") != "token_count":
-            continue
-          info = payload.get("info") or {}
-          # total_token_usage is cumulative for the session. Adding every
-          # snapshot makes usage grow quadratically, so count the last turn.
-          usage = info.get("last_token_usage") or {}
-          cache_read = number(usage.get("cached_input_tokens"))
-          cache_write = number(usage.get("cache_write_input_tokens"))
-          # Cached tokens are included in input_tokens, and reasoning tokens
-          # are included in output_tokens. Keep the cache split without
-          # counting either category twice.
-          input_tokens = max(0, number(usage.get("input_tokens")) - cache_read - cache_write)
-          output_tokens = number(usage.get("output_tokens"))
-          if not (input_tokens or output_tokens or cache_read or cache_write):
-            continue
-          day = local_day(entry.get("timestamp") or path.stat().st_mtime)
-          add_usage(day, str(path), current_model, input_tokens, output_tokens, cache_read, cache_write)
-    except Exception:
-      continue
+          previous_total_usage = total_usage
+        day = local_day(entry.get("timestamp") or stat_result.st_mtime)
+        record_usage(record, day, current_model, input_tokens, output_tokens, cache_read, cache_write)
+  except Exception:
+    # Keep what was read before the failure, but caching a partial record would
+    # hide the rest until the file changes, so hand back one the cache refuses.
+    record["mtime"] = None
+  return record
 
 
 def cache_root():
@@ -343,16 +542,86 @@ def cache_root():
   return root
 
 
-def scan_cache_paths():
+def scan_cache_digest():
   codex_home = Path(os.environ.get("CODEX_HOME") or (Path.home() / ".codex"))
   db = Path(os.environ.get("XDG_DATA_HOME") or (Path.home() / ".local" / "share")) / "opencode" / "opencode.db"
   # The digest covers every data path the scan reads: the codex session
   # roots, the opencode DB, and (via Path.home()) the pi/omp session roots.
-  digest = hashlib.sha1((str(Path.home()) + "\n" + str(codex_home) + "\n" + str(db)).encode("utf-8")).hexdigest()[:16]
+  return hashlib.sha1((str(Path.home()) + "\n" + str(codex_home) + "\n" + str(db)).encode("utf-8")).hexdigest()[:16]
+
+
+def scan_cache_paths():
+  digest = scan_cache_digest()
   root = cache_root()
   return root / f"codex-scan-{digest}.json", root / f"codex-scan-{digest}.lock"
 
 
+# Records hold local days, so they are only true in the timezone that wrote them.
+def file_cache_zone():
+  return [os.environ.get("TZ"), list(time.tzname), time.timezone, time.altzone]
+
+
+def file_cache_path():
+  return cache_root() / f"codex-files-{scan_cache_digest()}.json"
+
+
+def valid_file_record(value):
+  try:
+    counts = [totals for models in value["days"].values() for totals in models.values()]
+    return (
+      type(value["mtime"]) in (int, float) and type(value["size"]) is int
+      and all(len(totals) == 5 and all(type(n) is int for n in totals) for totals in counts)
+    )
+  except Exception:
+    return False
+
+
+def read_file_cache():
+  """Per-file totals of native sessions from earlier scans, keyed by path.
+
+  Session files are append-only, so a file whose mtime and size both match the
+  record cannot have grown a turn since: its totals can be replayed instead of
+  read. That is what keeps a refresh off the gigabytes of history it already
+  counted. Unlike the aggregate scan cache these records carry no "today" and
+  never expire on their own -- only the file they describe or a change of
+  timezone can invalidate one, so midnight and the clock moving leave them true.
+
+  pi and omp sessions stay out of it: a forked pi session repeats its parent's
+  messages, so they are only deduplicated across the whole tree at once.
+  """
+  try:
+    payload = json.loads(file_cache_path().read_text(encoding="utf-8"))
+  except Exception:
+    return {}
+  if not isinstance(payload, dict) or payload.get("schemaVersion") != FILE_CACHE_SCHEMA:
+    return {}
+  if payload.get("zone") != file_cache_zone():
+    return {}
+  files = payload.get("files")
+  if not isinstance(files, dict):
+    return {}
+  return {path: record for path, record in files.items() if valid_file_record(record)}
+
+
+def write_file_cache(files):
+  # Only the files this scan saw are written back, so deleted sessions and
+  # sessions that aged out of the 30-day window leave the cache instead of
+  # growing it forever.
+  try:
+    write_json(file_cache_path(), {"schemaVersion": FILE_CACHE_SCHEMA, "zone": file_cache_zone(), "files": files})
+  except Exception as exc:
+    print(f"omarchy-agent-usage-codex: could not write file cache ({exc})", file=sys.stderr)
+
+
+def reusable_record(cached_files, key, stat_result):
+  record = cached_files.get(key)
+  if record is None:
+    return None
+  if record["mtime"] != stat_result.st_mtime or record["size"] != stat_result.st_size:
+    return None
+  return record
+
+
 def read_fresh_json(path, max_age_seconds):
   if max_age_seconds <= 0 or not path.exists():
     return None
@@ -389,9 +658,10 @@ def write_json(path, payload):
 # The cache payload is a versioned envelope around the local-stats dict, so a
 # corrupted or foreign-shaped file is a cache miss (rescan + rewrite) instead
 # of a crash or a garbage record.
+# Version 2 invalidates totals counted before native notification deduplication.
 def read_cached_stats(cache_file, max_age_seconds):
   cached = read_fresh_json(cache_file, max_age_seconds)
-  if not isinstance(cached, dict) or cached.get("schemaVersion") != 1:
+  if not isinstance(cached, dict) or cached.get("schemaVersion") != 2:
     return None
   # today* fields only mean "today" on the day they were scanned. A cache
   # from another local date (midnight passed, or the clock moved) is a miss,
@@ -408,7 +678,7 @@ def read_cached_stats(cache_file, max_age_seconds):
 
 def write_cached_stats(cache_file, stats):
   try:
-    write_json(cache_file, {"schemaVersion": 1, "scanDate": today, "stats": stats})
+    write_json(cache_file, {"schemaVersion": 2, "scanDate": today, "stats": stats})
   except Exception as exc:
     print(f"omarchy-agent-usage-codex: could not write usage cache ({exc})", file=sys.stderr)
 
@@ -432,10 +702,14 @@ def local_stats():
   }
 
 
-def run_local_scans():
+# A forced scan (no reuse at all) reads every session file again.
+def run_local_scans(max_age):
+  cached_files = read_file_cache() if max_age > 0 else {}
+  scanned_files = {}
   scan_pi_sessions()
-  scan_native_codex_sessions()
+  scan_native_codex_sessions(cached_files, scanned_files)
   complete = scan_opencode_sessions()
+  write_file_cache(scanned_files)
   return local_stats(), complete
 
 
@@ -450,7 +724,7 @@ def cached_local_stats(max_age):
     return _cached_local_stats(max_age)
   except Exception as exc:
     print(f"omarchy-agent-usage-codex: cache unavailable ({exc}); scanning directly", file=sys.stderr)
-    stats, _ = run_local_scans()
+    stats, _ = run_local_scans(max_age)
     return stats
 
 
@@ -466,7 +740,7 @@ def _cached_local_stats(max_age):
     cached = read_cached_stats(cache_file, max_age)
     if cached is not None:
       return cached
-    stats, complete = run_local_scans()
+    stats, complete = run_local_scans(max_age)
     # An interrupted scan still serves this run, but caching it would
     # suppress the missing usage for every reader until the cache expires.
     if complete:
@@ -474,25 +748,55 @@ def _cached_local_stats(max_age):
     return stats
 
 
+def rpc_send(proc, payload, operation):
+  """Write one JSON-RPC frame. Raise immediately if the app-server is gone."""
+  try:
+    proc.stdin.write(json.dumps(payload) + "\n")
+    proc.stdin.flush()
+  except OSError as exc:
+    try:
+      proc.stdin.close()
+    except OSError:
+      pass
+    raise RuntimeError(f"Codex app-server exited before {operation}") from exc
+
+
 def rpc_request(proc, request_id, method, params=None, timeout=8):
   payload = {"id": request_id, "method": method, "params": params or {}}
-  proc.stdin.write(json.dumps(payload) + "\n")
-  proc.stdin.flush()
-  deadline = time.time() + timeout
-  while time.time() < deadline:
-    ready, _, _ = select.select([proc.stdout], [], [], 0.25)
-    if not ready:
-      continue
-    line = proc.stdout.readline()
-    if not line:
-      break
-    try:
-      message = json.loads(line)
-    except Exception:
-      continue
-    if message.get("id") == request_id:
-      return message
-  raise TimeoutError(method)
+  rpc_send(proc, payload, method)
+  # Read the raw fd and split lines ourselves. The app-server can emit
+  # notifications in the same write as a reply; a buffered readline() would
+  # pull them all into Python's buffer where select() cannot see them, and
+  # the reply would sit there until the deadline.
+  fd = proc.stdout.fileno()
+  pending = getattr(proc, "_rpc_pending", b"")
+  deadline = time.monotonic() + timeout
+  try:
+    while True:
+      while b"\n" in pending:
+        line, pending = pending.split(b"\n", 1)
+        try:
+          message = json.loads(line)
+        except Exception:
+          continue
+        if isinstance(message, dict) and message.get("id") == request_id:
+          return message
+      remaining = deadline - time.monotonic()
+      if remaining <= 0:
+        break
+      ready, _, _ = select.select([fd], [], [], min(0.25, remaining))
+      if not ready:
+        if proc.poll() is not None:
+          raise RuntimeError(f"Codex app-server exited before {method}")
+        continue
+      chunk = os.read(fd, 65536)
+      if not chunk:
+        raise RuntimeError(f"Codex app-server exited before {method}")
+      pending += chunk
+  finally:
+    proc._rpc_pending = pending
+  # Process still up but silent: keep the method name searchable, but name the stall.
+  raise TimeoutError(f"Codex app-server did not answer {method}")
 
 
 def limit_window(window):
@@ -518,6 +822,33 @@ def limit_window(window):
   }
 
 
+def codex_rpc_help(proc, stderr_file, exc):
+  """Explain an app-server RPC failure: a live server's stall, a dead one's own
+  stderr, or the login hint when it died saying nothing."""
+  try:
+    proc.wait(timeout=1)
+  except Exception:
+    pass
+
+  text = str(exc).strip()
+  if proc.poll() is None:
+    # Still up: stall or unparseable payload. Not an auth problem, and any
+    # stderr so far is its logging, not why it stopped.
+    return text[:300] or AUTH_HELP
+
+  try:
+    stderr_file.seek(0)
+    detail = stderr_file.read()
+  except Exception:
+    detail = ""
+  lines = [line.strip() for line in detail.splitlines() if line.strip()]
+  if lines:
+    # The fatal error comes last, after whatever the CLI logged on the way up.
+    return f"codex app-server exited: {' '.join(lines)[-275:]}"
+  # Exited cleanly with empty stderr — typical "not logged in" shape.
+  return AUTH_HELP
+
+
 def reset_credits(result):
   granted = (result.get("rateLimitResetCredits") or {}).get("credits") or []
   available = [c for c in granted if isinstance(c, dict) and c.get("status") == "available"]
@@ -530,33 +861,70 @@ def reset_credits(result):
   }
 
 
+# Keyring and auto stores keep credentials outside auth.json, so only the file
+# store can be known to be empty without asking Codex.
+def has_codex_credentials(codex_home, env):
+  if env.get("CODEX_ACCESS_TOKEN"):
+    return True
+  try:
+    with (codex_home / "config.toml").open("rb") as handle:
+      config = tomllib.load(handle)
+  except FileNotFoundError:
+    config = {}
+  except Exception:
+    # A config this can't read may still name another store; let Codex say.
+    return True
+  if str(config.get("cli_auth_credentials_store") or "file") != "file":
+    return True
+  return (codex_home / "auth.json").is_file()
+
+
 def fetch_codex_rpc(home=None):
   result = {"limits": [], "tierLabel": "", "usageStatusText": "", "authHelpText": AUTH_HELP}
   env = dict(ENV, CODEX_HOME=str(home)) if home else ENV
-  codex = find_command("codex")
+  codex = find_codex_binary()
   if not codex:
     result["usageStatusText"] = "Codex unavailable"
     result["authHelpText"] = "codex not found in PATH"
     return result
 
+  # No auth.json under the file store and no CODEX_ACCESS_TOKEN means
+  # account/read can only fail — and starting the app-server is not free: it
+  # syncs the plugin list, a git fetch per refresh that also leaves
+  # .tmp/git-* folders behind. The home is the one being probed, so a
+  # signed-out secondary account reads as signed out even while ~/.codex
+  # holds a login.
+  if not has_codex_credentials(Path(env.get("CODEX_HOME") or (Path.home() / ".codex")), env):
+    result["usageStatusText"] = "Waiting for auth"
+    result["authHelpText"] = AUTH_HELP
+    return result
+
+  # stderr goes to a regular file, not a PIPE: an unread pipe can fill and
+  # deadlock a chatty app-server. It is only read once the process has exited.
+  # Without temporary space the probe still runs, just without its error text.
+  try:
+    stderr_file = tempfile.TemporaryFile("w+", errors="replace")
+  except OSError:
+    stderr_file = None
   try:
     proc = subprocess.Popen(
       [codex, "-s", "read-only", "-a", "on-request", "app-server"],
       stdin=subprocess.PIPE,
       stdout=subprocess.PIPE,
-      stderr=subprocess.DEVNULL,
+      stderr=stderr_file or subprocess.DEVNULL,
       text=True,
       env=env,
     )
   except Exception as exc:
+    if stderr_file:
+      stderr_file.close()
     result["usageStatusText"] = "Codex unavailable"
     result["authHelpText"] = str(exc)
     return result
 
   try:
     rpc_request(proc, 1, "initialize", {"clientInfo": {"name": "omarchy-agent-usage", "version": "1"}}, timeout=8)
-    proc.stdin.write(json.dumps({"method": "initialized", "params": {}}) + "\n")
-    proc.stdin.flush()
+    rpc_send(proc, {"method": "initialized", "params": {}}, "initialized")
     limits_msg = rpc_request(proc, 2, "account/rateLimits/read", timeout=8)
 
     # A home nobody is signed in to answers with an error rather than limits.
@@ -573,6 +941,7 @@ def fetch_codex_rpc(home=None):
         result["authHelpText"] = message
       return result
     limits = (limits_msg.get("result") or {}).get("rateLimits") or {}
+    result["authHelpText"] = ""
     plan = limits.get("planType") or ""
 
     # The limits name the plan themselves. account/read is only a fallback
@@ -583,7 +952,7 @@ def fetch_codex_rpc(home=None):
         account_msg = rpc_request(proc, 3, "account/read", timeout=2)
         account = (account_msg.get("result") or {}).get("account") or {}
         plan = account.get("planType") or account.get("type") or ""
-      except TimeoutError:
+      except (TimeoutError, RuntimeError):
         pass
     result["tierLabel"] = str(plan) if plan else ""
 
@@ -599,7 +968,7 @@ def fetch_codex_rpc(home=None):
       result["resetCredits"] = credits
   except Exception as exc:
     result["usageStatusText"] = "Codex limits unavailable"
-    result["authHelpText"] = str(exc)
+    result["authHelpText"] = codex_rpc_help(proc, stderr_file, exc)
   finally:
     try:
       proc.terminate()
@@ -609,6 +978,8 @@ def fetch_codex_rpc(home=None):
         proc.kill()
       except Exception:
         pass
+    if stderr_file:
+      stderr_file.close()
   return result
 
 

bin/omarchy-agent-usage-fireworks

@@ -29,7 +29,7 @@ from typing import Any
 
 AGENT_ID = "fireworks"
 AGENT_NAME = "Fireworks"
-AUTH_HELP = "Set FIREWORKS_API_KEY, run `firectl set-api-key`, or sign in to Fireworks in opencode."
+AUTH_HELP = "Set FIREWORKS_API_KEY, run `firectl set-api-key`, or sign in to Fireworks in pi or opencode."
 API_BASE_URL = "https://api.fireworks.ai"
 
 
@@ -196,11 +196,57 @@ def read_auth_file(path: Path) -> tuple[str, str]:
   return api_key, account_id
 
 
+def pi_auth_path() -> Path:
+  agent_dir = Path(os.environ.get("PI_CODING_AGENT_DIR") or (Path.home() / ".pi" / "agent"))
+  return agent_dir / "auth.json"
+
+
 def opencode_auth_path() -> Path:
   data_home = Path(os.environ.get("XDG_DATA_HOME") or (Path.home() / ".local" / "share"))
   return data_home / "opencode" / "auth.json"
 
 
+PI_ENV_REFERENCE = re.compile(r"\$(?:\{([A-Za-z_][A-Za-z0-9_]*)\}|([A-Za-z_][A-Za-z0-9_]*))")
+
+
+def resolve_pi_key(raw: Any) -> str:
+  # pi credential keys are literals, $ENV_VAR/${ENV_VAR} references, or
+  # !command shell lookups. The collector resolves the first two; command
+  # lookups stay pi-only and are skipped rather than sent to the API verbatim.
+  key = str(raw or "").strip()
+  if not key or key.startswith("!"):
+    return ""
+  # Escapes survive interpolation as sentinels, then restore.
+  key = key.replace("$$", "\x00").replace("$!", "\x01")
+
+  unresolved = False
+
+  def substitute(match: re.Match[str]) -> str:
+    nonlocal unresolved
+    value = os.environ.get(match.group(1) or match.group(2), "")
+    if not value:
+      unresolved = True
+    return value
+
+  key = PI_ENV_REFERENCE.sub(substitute, key)
+  if unresolved:
+    return ""
+  return key.replace("\x00", "$").replace("\x01", "!")
+
+
+def read_pi_key(path: Path) -> str:
+  try:
+    parsed = json.loads(path.read_text())
+  except (OSError, json.JSONDecodeError):
+    return ""
+  entry = parsed.get("fireworks") if isinstance(parsed, dict) else None
+  if not isinstance(entry, dict):
+    return ""
+  if entry.get("type") not in (None, "api_key"):
+    return ""
+  return resolve_pi_key(entry.get("key"))
+
+
 def read_opencode_key(path: Path) -> str:
   try:
     parsed = json.loads(path.read_text())
@@ -227,11 +273,12 @@ def read_config() -> dict[str, Any]:
 
 def credentials(auth_path: Path, config: dict[str, Any]) -> tuple[str, str]:
   file_key, file_account = read_auth_file(auth_path)
-  # opencode is the last resort: an explicit key or a firectl login should
-  # win over whatever another tool happens to be signed in with.
+  # Agent logins are the fallback ladder below an explicit key or a firectl
+  # login: pi first as Omarchy's default agent, opencode as the last resort.
   api_key = (
     str(os.environ.get("FIREWORKS_API_KEY", "")).strip()
     or file_key
+    or read_pi_key(pi_auth_path())
     or read_opencode_key(opencode_auth_path())
   )
   account_id = (

bin/omarchy-agent-usage-grok

@@ -30,6 +30,9 @@ CREDITS_URL = "https://cli-chat-proxy.grok.com/v1/billing?format=credits"
 # Grok's own CLI watches its subscription once a minute; a panel opened and
 # shut repeatedly reuses an answer for about that long.
 PROBE_REUSE_SECONDS = 60
+# Grok's docs give a saved sign-in 30 days; a refresh token older than that
+# can't renew the access token, so it no longer counts as signed in.
+REFRESH_LIFETIME_SECONDS = 30 * 24 * 3600
 # A limits-only refresh needs fresh limits, not a fresh history scan.
 LIMITS_ONLY_SCAN_REUSE_SECONDS = 900
 
@@ -190,6 +193,22 @@ def credits_limit(data):
   }
 
 
+# A window whose reset has passed starts over empty. A weekly one's next
+# reset is a whole number of weeks later; a monthly one's isn't known.
+def rolled_forward(limits):
+  now = datetime.now(timezone.utc)
+  rolled = []
+  for limit in limits:
+    resets = parse_time(limit.get("resetsAt"))
+    if resets and resets <= now:
+      limit = dict(limit, percent=0.0, resetsAt="")
+      if limit.get("label") == "Weekly":
+        weeks = (now - resets) // timedelta(weeks=1) + 1
+        limit["resetsAt"] = (resets + weeks * timedelta(weeks=1)).isoformat()
+    rolled.append(limit)
+  return rolled
+
+
 def collect_limits(home, key, force):
   result = {"limits": [], "usageStatusText": "", "authHelpText": AUTH_HELP, "live": False, "fetchedAtMs": 0}
   entry = login(home)
@@ -203,17 +222,25 @@ def collect_limits(home, key, force):
   fetched_at = float(cached.get("fetchedAtMs") or 0)
   result["fetchedAtMs"] = fetched_at
 
-  # The CLI refreshes its token while it runs; one left to lapse can't be
-  # used until Grok starts again.
   expires = parse_time(entry.get("expires_at"))
   if expires and expires.timestamp() <= time.time():
     result["limits"] = cached.get("limits") or []
-    result["usageStatusText"] = "Sign-in expired"
-    result["authHelpText"] = "Grok's saved sign-in expired. Start Grok to refresh it."
+    # The access token lives six hours and Grok mints a new one from its
+    # refresh token whenever it starts, so a lapsed one is routine, not a
+    # sign-out. And while Grok hasn't run, nothing here has spent its
+    # allowance, so the last numbers still stand, rolled past any reset.
+    if entry.get("refresh_token") and time.time() - expires.timestamp() < REFRESH_LIFETIME_SECONDS:
+      result.update(limits=rolled_forward(result["limits"]), live=True)
+      if not result["limits"]:
+        result["usageStatusText"] = "Limits paused"
+        result["authHelpText"] = "Start Grok to see its limits; signing in again is not needed."
+    else:
+      result["usageStatusText"] = "Sign-in expired"
+      result["authHelpText"] = "Grok's saved sign-in expired. Start Grok to refresh it."
     return result
 
   if cached.get("limits") and not force and time.time() - fetched_at / 1000 < PROBE_REUSE_SECONDS:
-    result.update(limits=cached["limits"], live=True)
+    result.update(limits=cached["limits"], live=True, authHelpText="")
     return result
 
   request = urllib.request.Request(CREDITS_URL, headers={"Authorization": f"Bearer {token}", "Accept": "application/json"})
@@ -228,7 +255,7 @@ def collect_limits(home, key, force):
     limit = None
 
   if limit:
-    result.update(limits=[limit], live=True, fetchedAtMs=round(time.time() * 1000))
+    result.update(limits=[limit], live=True, fetchedAtMs=round(time.time() * 1000), authHelpText="")
     write_json(cache, {"fetchedAtMs": result["fetchedAtMs"], "limits": result["limits"]})
   else:
     result["limits"] = cached.get("limits") or []