931 lines
34 KiB
Python
Executable File
931 lines
34 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Claude Code status line.
|
|
|
|
One line: vim mode | dir (repo-relative, hidden at the root) |
|
|
git(repo/branch +dirty +ahead/behind) | PR review threads (open count, Codex's share) | model (+effort) |
|
|
context% (+compact warn) | 5h usage (+reset eta) | cost (+burn rate) |
|
|
session tokens (cumulative in/out) | decode speed t/s (thinking blocks) |
|
|
prompt-cache hit% (+ttl, cold marker) | disk. If the
|
|
rendered line would overflow the terminal, trailing (lowest-priority) segments are
|
|
dropped until it fits.
|
|
(Dormant, re-addable in main(): ram, cpu%, temp -- retired 2026-07-26 because the
|
|
status line then only refreshed on message events, so system metrics sat visibly stale
|
|
(settings.json's 60 s refreshInterval, added for the PR count, is still too coarse for them);
|
|
tmux's status bar owns those now -- plus velocity, cache hit %, api ratio, version,
|
|
style.)
|
|
|
|
Reads the status JSON from stdin (Claude Code statusLine command). Every segment
|
|
is wrapped defensively so the status line can never crash the UI -- on any error
|
|
a segment is simply omitted.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import unicodedata
|
|
|
|
# --- ANSI helpers -----------------------------------------------------------
|
|
RESET = "\033[0m"
|
|
DIM = "" # disabled: "\033[2m" blended the gray detail text into dark backgrounds
|
|
BOLD = "\033[1m"
|
|
C = {
|
|
"cyan": "\033[36m",
|
|
"green": "\033[32m",
|
|
"yellow": "\033[33m",
|
|
"red": "\033[31m",
|
|
"magenta": "\033[35m",
|
|
"blue": "\033[34m",
|
|
"gray": "\033[37m", # light gray (was "\033[90m" bright-black, too dark to read)
|
|
}
|
|
|
|
|
|
def color(text, name="", dim=False, bold=False):
|
|
pre = ""
|
|
if dim:
|
|
pre += DIM
|
|
if bold:
|
|
pre += BOLD
|
|
pre += C.get(name, "")
|
|
return f"{pre}{text}{RESET}" if pre else str(text)
|
|
|
|
|
|
def bucket(value, lo, hi, invert=False):
|
|
"""green/yellow/red by threshold. invert=True -> high is good."""
|
|
if invert:
|
|
return "green" if value >= hi else ("yellow" if value >= lo else "red")
|
|
return "green" if value < lo else ("yellow" if value < hi else "red")
|
|
|
|
|
|
DIV = color(" │ ", "gray", dim=True)
|
|
|
|
|
|
def join_line(segs):
|
|
return DIV.join(s for s in segs if s)
|
|
|
|
|
|
_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")
|
|
|
|
|
|
def vlen(text):
|
|
"""Visible width of a rendered string: ANSI stripped, emoji counted as 2."""
|
|
text = _ANSI_RE.sub("", text)
|
|
w = 0
|
|
for ch in text:
|
|
if ch in ("\u200d", "\ufe0f", "\ufe0e") or unicodedata.combining(ch):
|
|
continue # ZWJ / variation selectors / combining marks: zero width
|
|
o = ord(ch)
|
|
if o >= 0x1F000 or 0x2600 <= o <= 0x27BF or unicodedata.east_asian_width(ch) in ("W", "F"):
|
|
w += 2
|
|
else:
|
|
w += 1
|
|
return w
|
|
|
|
|
|
def term_cols():
|
|
"""Terminal width. Claude Code exports COLUMNS; 0 if unknown -> stay one line."""
|
|
try:
|
|
return shutil.get_terminal_size(fallback=(0, 0)).columns
|
|
except Exception:
|
|
return 0
|
|
|
|
|
|
# --- context usage ------------------------------------------------------------
|
|
_USAGE_KEYS = ("input_tokens", "cache_creation_input_tokens", "cache_read_input_tokens")
|
|
|
|
|
|
def payload_usage(data):
|
|
"""`context_window.current_usage` from the status payload, shaped like `message.usage`.
|
|
|
|
Claude Code 2.1.x hands the status line its own token accounting, so the transcript
|
|
scan below is only a fallback for payloads that predate it (or lack a transcript).
|
|
"""
|
|
cw = data.get("context_window")
|
|
cu = cw.get("current_usage") if isinstance(cw, dict) else None
|
|
if isinstance(cu, dict) and any(isinstance(cu.get(k), (int, float)) for k in _USAGE_KEYS):
|
|
return cu
|
|
return {}
|
|
|
|
|
|
# --- shared transcript usage (fallback) ------------------------------------------
|
|
def _scan_usage(text):
|
|
"""Last `message.usage` block among these JSONL lines, or {}."""
|
|
usage = {}
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if not line or '"usage"' not in line:
|
|
continue
|
|
try:
|
|
obj = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
msg = obj.get("message")
|
|
if isinstance(msg, dict) and isinstance(msg.get("usage"), dict):
|
|
usage = msg["usage"]
|
|
return usage
|
|
|
|
|
|
TRANSCRIPT_TAIL = 1 << 20 # 1 MiB
|
|
|
|
|
|
def last_usage(data):
|
|
"""Most recent `message.usage` block from the transcript, or {}.
|
|
|
|
Fallback only -- see payload_usage(). Reads only the tail: this runs on every status refresh and transcripts grow to
|
|
tens of MB, so re-parsing the whole file each time is unbounded work for a value
|
|
that lives at the end. Falls back to a full scan if the tail holds no usage block
|
|
(a long stretch of tool results can push it past the window).
|
|
"""
|
|
try:
|
|
path = data.get("transcript_path")
|
|
if not path or not os.path.exists(path):
|
|
return {}
|
|
with open(path, "rb") as fh:
|
|
size = fh.seek(0, os.SEEK_END)
|
|
start = max(0, size - TRANSCRIPT_TAIL)
|
|
fh.seek(start)
|
|
text = fh.read().decode("utf-8", "ignore")
|
|
if start:
|
|
text = text.partition("\n")[2] # drop the partial first line
|
|
usage = _scan_usage(text)
|
|
if not usage and start:
|
|
with open(path, "r", errors="ignore") as fh:
|
|
usage = _scan_usage(fh.read())
|
|
return usage
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
# --- session token totals (transcript, incremental) ---------------------------
|
|
TOKENS_STATE_DIR = os.path.expanduser("~/.claude/cache")
|
|
TOKENS_RECENT_IDS = 20 # message ids remembered across runs for de-duplication
|
|
TPS_SAMPLES = 8 # thinking blocks the t/s readout is averaged over
|
|
TPS_MIN_MS, TPS_MIN_TOKENS = 1000, 50 # shorter blocks are first-token latency, not decode speed
|
|
TOKENS_SCAN_BUDGET = 8 << 20 # bytes parsed per refresh (~0.1 s); see session_tokens
|
|
TOKENS_STATE_MAX_AGE = 7 * 86400 # prune per-session state files older than this
|
|
|
|
|
|
def fmt_tokens(n):
|
|
"""1234 -> 1.2k, 843000 -> 843k, 1_234_567 -> 1.2M, 12_345_678 -> 12M.
|
|
|
|
Thresholds sit at the rounding boundaries (0.9995, 9.95) so a value never prints as
|
|
"10.0k" or "1000k" -- it rolls into the shorter form / the next unit instead."""
|
|
n = float(n)
|
|
for unit, div in (("M", 1e6), ("k", 1e3)):
|
|
v = n / div
|
|
if v < 0.9995:
|
|
continue
|
|
return f"{v:.0f}{unit}" if v >= 9.95 else f"{v:.1f}{unit}"
|
|
return f"{n:.0f}"
|
|
|
|
|
|
def _tokens_state_path(data, path):
|
|
sid = data.get("session_id") or os.path.splitext(os.path.basename(path))[0]
|
|
sid = re.sub(r"[^A-Za-z0-9_-]", "_", str(sid))[:80]
|
|
return os.path.join(TOKENS_STATE_DIR, f"statusline.tokens.{sid}.json")
|
|
|
|
|
|
def _prune_tokens_state(keep):
|
|
"""Drop stale per-session token and per-branch PR state files (one scandir, only when
|
|
the dir has piled up)."""
|
|
try:
|
|
entries = [e for e in os.scandir(TOKENS_STATE_DIR) if e.name.startswith(("statusline.tokens.", "statusline.pr."))]
|
|
if len(entries) < 50:
|
|
return
|
|
cutoff = time.time() - TOKENS_STATE_MAX_AGE
|
|
for e in entries:
|
|
if e.path != keep and e.stat().st_mtime < cutoff:
|
|
os.unlink(e.path)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def session_tokens(data):
|
|
"""Cumulative session token totals plus decode-speed samples, from the transcript --
|
|
`{"in", "out", "tps"}` or None when there is no transcript.
|
|
|
|
in/out: summed over assistant `message.usage` blocks. Input = input + cache_creation +
|
|
cache_read (everything the API billed as input), so on a 1M session it dwarfs output:
|
|
every call re-reads the whole cached prefix. Only the main transcript is counted;
|
|
subagent usage lives in separate sidechain files.
|
|
|
|
tps: the last TPS_SAMPLES `[thinking_tokens, thinkingDurationMs]` pairs -- the only
|
|
clean per-call timing Claude Code records. The status payload's cost.total_api_duration_ms
|
|
is useless for a rate: subagent and advisor calls land in the same counter (it grew 28s
|
|
in a 12s wall interval with no main-conversation completion), and the payload's
|
|
output_tokens is sometimes a mid-stream partial. thinkingDurationMs is main-conversation
|
|
only; it includes first-token latency, so short blocks read low and long ones converge
|
|
on the real decode speed -- hence the duration-weighted mean and the minimum-size filter.
|
|
|
|
Incremental: the status line runs on every refresh and transcripts grow to tens of
|
|
MB, so a per-session state file (~/.claude/cache/statusline.tokens.<session>.json)
|
|
remembers the byte offset already summed and only the new tail is parsed. Lines are
|
|
de-duplicated by message id: the transcript writes one line per content block of a
|
|
turn, all carrying the same id and the same usage block. Only complete lines are
|
|
consumed (the writer may be mid-line); a shrunken file (truncate/rotate) resets.
|
|
|
|
A cold scan is spread over refreshes, TOKENS_SCAN_BUDGET bytes each: Claude Code kills
|
|
a status script when the next refresh fires, and a multi-second scan of a 100 MB
|
|
transcript that never gets to write its state would start from zero every time.
|
|
Totals climb for a few refreshes on a huge resumed transcript, then settle.
|
|
"""
|
|
path = data.get("transcript_path")
|
|
if not path or not os.path.exists(path):
|
|
return None
|
|
state_path = _tokens_state_path(data, path)
|
|
fresh = {"path": path, "offset": 0, "in": 0, "out": 0, "ids": [], "tps": [], "tps_ids": []}
|
|
state = dict(fresh)
|
|
try:
|
|
with open(state_path) as fh:
|
|
s = json.load(fh)
|
|
if (
|
|
isinstance(s, dict)
|
|
and s.get("path") == path
|
|
and all(isinstance(s.get(k), int) and s[k] >= 0 for k in ("offset", "in", "out"))
|
|
and isinstance(s.get("ids"), list)
|
|
):
|
|
state = s # "tps"/"tps_ids" are optional: older state files predate them
|
|
except Exception:
|
|
pass
|
|
|
|
def result(st):
|
|
return {"in": st["in"], "out": st["out"], "tps": st.get("tps") or []}
|
|
|
|
size = os.path.getsize(path)
|
|
if size < state["offset"]: # truncated or rotated under us: start over
|
|
state = dict(fresh)
|
|
if size == state["offset"]:
|
|
return result(state)
|
|
|
|
with open(path, "rb") as fh:
|
|
fh.seek(state["offset"])
|
|
chunk = fh.read(TOKENS_SCAN_BUDGET)
|
|
end = chunk.rfind(b"\n")
|
|
if end < 0 and len(chunk) == TOKENS_SCAN_BUDGET: # one line bigger than the budget: finish it
|
|
chunk += fh.readline()
|
|
end = chunk.rfind(b"\n")
|
|
if end < 0: # no complete new line yet
|
|
return result(state)
|
|
text = chunk[: end + 1].decode("utf-8", "ignore")
|
|
|
|
tin, tout = state["in"], state["out"]
|
|
ids = [i for i in state["ids"] if isinstance(i, str)]
|
|
seen = set(ids)
|
|
tps = [s for s in (state.get("tps") or []) if isinstance(s, list) and len(s) == 2]
|
|
tps_ids = [i for i in (state.get("tps_ids") or []) if isinstance(i, str)]
|
|
tps_seen = set(tps_ids)
|
|
for line in text.splitlines():
|
|
if '"usage"' not in line:
|
|
continue
|
|
try:
|
|
obj = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
msg = obj.get("message") if isinstance(obj, dict) else None
|
|
usage = msg.get("usage") if isinstance(msg, dict) else None
|
|
if not isinstance(usage, dict):
|
|
continue
|
|
mid = msg.get("id") or obj.get("requestId") or obj.get("uuid")
|
|
if not isinstance(mid, str):
|
|
continue
|
|
|
|
def num(d, k):
|
|
v = d.get(k, 0) if isinstance(d, dict) else 0
|
|
return int(v) if isinstance(v, (int, float)) else 0
|
|
|
|
# Decode-speed sample: checked on every line (not just the first-wins one below) so
|
|
# it does not depend on the thinking block being the message's first line.
|
|
ms = num(obj, "thinkingDurationMs")
|
|
toks = num(usage.get("output_tokens_details"), "thinking_tokens")
|
|
if ms >= TPS_MIN_MS and toks >= TPS_MIN_TOKENS and mid not in tps_seen:
|
|
tps_seen.add(mid)
|
|
tps_ids.append(mid)
|
|
tps.append([toks, ms])
|
|
|
|
if mid in seen:
|
|
continue
|
|
seen.add(mid)
|
|
ids.append(mid)
|
|
tin += num(usage, "input_tokens") + num(usage, "cache_creation_input_tokens") + num(usage, "cache_read_input_tokens")
|
|
tout += num(usage, "output_tokens")
|
|
|
|
state = {
|
|
"path": path,
|
|
"offset": state["offset"] + end + 1,
|
|
"in": tin,
|
|
"out": tout,
|
|
"ids": ids[-TOKENS_RECENT_IDS:],
|
|
"tps": tps[-TPS_SAMPLES:],
|
|
"tps_ids": tps_ids[-TOKENS_RECENT_IDS:],
|
|
}
|
|
try:
|
|
os.makedirs(TOKENS_STATE_DIR, exist_ok=True)
|
|
tmp = f"{state_path}.{os.getpid()}.tmp"
|
|
with open(tmp, "w") as fh:
|
|
json.dump(state, fh)
|
|
os.replace(tmp, state_path) # atomic: a concurrent refresh never reads a half-written file
|
|
_prune_tokens_state(state_path)
|
|
except Exception:
|
|
pass
|
|
return result(state)
|
|
|
|
|
|
# --- row 1: work ------------------------------------------------------------
|
|
def vim_segment(data):
|
|
"""Vim editor mode as a compact [I]/[N]/[V] (only present with editorMode: vim).
|
|
|
|
Pair with `"hideVimModeIndicator": true` in the statusLine settings block so
|
|
Claude's own built-in indicator doesn't duplicate this one.
|
|
"""
|
|
mode = (data.get("vim") or {}).get("mode")
|
|
if not mode:
|
|
return ""
|
|
m = mode.upper()
|
|
if m.startswith("INSERT"):
|
|
return color("[I]", "green", bold=True)
|
|
if m.startswith("VISUAL"):
|
|
return color("[V]", "magenta", bold=True)
|
|
if m.startswith("NORMAL"):
|
|
return color("[N]", "blue", bold=True)
|
|
return color(f"[{m[:1]}]", "gray")
|
|
|
|
|
|
def abbrev(path):
|
|
"""fish-style prompt_pwd: shrink every component but the last to one char,
|
|
keeping a leading dot (`.config` -> `.c`)."""
|
|
parts = path.split(os.sep)
|
|
out = []
|
|
for i, p in enumerate(parts):
|
|
if i == len(parts) - 1 or p in ("", "~"):
|
|
out.append(p)
|
|
elif p.startswith("."):
|
|
out.append(p[:2])
|
|
else:
|
|
out.append(p[:1])
|
|
return os.sep.join(out)
|
|
|
|
|
|
def dir_segment(cwd, top=None):
|
|
"""Path *inside* the repo -- git_segment already names the repo itself, so
|
|
repeating the full ~-path here was pure duplication. Empty at the repo root;
|
|
a fish-shortened ~-path when we're not in a repo at all."""
|
|
try:
|
|
rel = os.path.relpath(cwd, top) if top else None
|
|
if rel and not rel.startswith(os.pardir): # `..` -> cwd is outside `top` (symlinked in); use the ~-path
|
|
if rel == os.curdir:
|
|
return ""
|
|
shown = rel if len(rel) <= 28 else abbrev(rel)
|
|
else:
|
|
home = os.path.expanduser("~")
|
|
shown = cwd
|
|
if cwd == home:
|
|
shown = "~"
|
|
elif cwd.startswith(home + os.sep):
|
|
shown = "~" + cwd[len(home):]
|
|
shown = abbrev(shown)
|
|
return "📁 " + color(shown, "cyan")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _git(cwd, *args):
|
|
return subprocess.run(
|
|
["git", "-C", cwd, *args], capture_output=True, text=True, timeout=1
|
|
)
|
|
|
|
|
|
def git_top(cwd):
|
|
"""Absolute path of the enclosing work tree, or None."""
|
|
try:
|
|
r = _git(cwd, "rev-parse", "--show-toplevel")
|
|
return r.stdout.strip() or None if r.returncode == 0 else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def git_segment(cwd, top):
|
|
try:
|
|
if not top:
|
|
return ""
|
|
repo = os.path.basename(top) or "repo"
|
|
branch = _git(cwd, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip() or "?"
|
|
|
|
# `status --porcelain` walks the whole work tree, so in a big repo it can blow
|
|
# the 1s timeout -- in its own try, because letting that escape dropped the repo
|
|
# and branch we already had: a decoration failing must not take the segment down.
|
|
try:
|
|
dirty = bool(_git(cwd, "status", "--porcelain").stdout.strip())
|
|
except Exception:
|
|
dirty = False
|
|
|
|
# ahead/behind vs upstream (omitted when no upstream is configured)
|
|
ab = ""
|
|
rl = _git(cwd, "rev-list", "--left-right", "--count", "@{upstream}...HEAD")
|
|
if rl.returncode == 0 and rl.stdout.strip():
|
|
try:
|
|
behind, ahead = (int(x) for x in rl.stdout.split())
|
|
bits = []
|
|
if ahead:
|
|
bits.append(color(f"↑{ahead}", "cyan"))
|
|
if behind:
|
|
bits.append(color(f"↓{behind}", "yellow"))
|
|
if bits:
|
|
ab = " " + "".join(bits)
|
|
except Exception:
|
|
ab = ""
|
|
|
|
mark = color("*", "red") if dirty else ""
|
|
label = color(repo, "magenta", bold=True)
|
|
br = color(branch, "yellow" if dirty else "green")
|
|
return f"🌿 {label} {br}{mark}{ab}"
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
# --- PR review threads (gh, cached, fetched in the background) ------------------
|
|
PR_TTL = 90 # seconds a fetched count stays fresh
|
|
PR_ERR_TTL = 600 # back-off after a failed fetch (no GitHub remote, gh not authed, offline)
|
|
PR_CLAIM = 30 # a refresher spawned this recently is still running: don't start another
|
|
PR_CODEX = re.compile(r"codex", re.I) # chatgpt-codex-connector; same match as /pr-loop
|
|
PR_QUERY = """query($owner:String!,$repo:String!,$branch:String!){
|
|
repository(owner:$owner,name:$repo){
|
|
pullRequests(headRefName:$branch,states:OPEN,first:1,orderBy:{field:UPDATED_AT,direction:DESC}){
|
|
nodes{ number reviewThreads(first:100){ pageInfo{hasNextPage}
|
|
nodes{ isResolved comments(first:1){ nodes{ author{login} } } } } } } } }"""
|
|
|
|
|
|
def _pr_cache_path(top, branch):
|
|
key = hashlib.sha1(f"{top}\0{branch}".encode()).hexdigest()[:16]
|
|
return os.path.join(TOKENS_STATE_DIR, f"statusline.pr.{key}.json")
|
|
|
|
|
|
def _read_json(path):
|
|
try:
|
|
with open(path) as fh:
|
|
d = json.load(fh)
|
|
return d if isinstance(d, dict) else {}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def pr_refresh(cwd, branch, path):
|
|
"""Background half of pr_segment (`statusline.py --pr-refresh CWD BRANCH CACHE`): fetch the
|
|
branch's open PR and write its unresolved review-thread counts to the cache. `{owner}` /
|
|
`{repo}` are gh placeholders resolved from the repo at `cwd`. A failed fetch keeps the last
|
|
counts and flags them stale; a branch without an open PR clears them."""
|
|
state = _read_json(path)
|
|
state["ts"] = time.time()
|
|
try:
|
|
r = subprocess.run(
|
|
["gh", "api", "graphql", "-F", "owner={owner}", "-F", "repo={repo}",
|
|
"-f", f"branch={branch}", "-f", f"query={PR_QUERY}"],
|
|
cwd=cwd, capture_output=True, text=True, timeout=20,
|
|
)
|
|
if r.returncode != 0:
|
|
raise RuntimeError(r.stderr)
|
|
nodes = json.loads(r.stdout)["data"]["repository"]["pullRequests"]["nodes"]
|
|
state = {"ts": state["ts"]}
|
|
if nodes:
|
|
threads = nodes[0]["reviewThreads"]
|
|
# a thread belongs to whoever opened it: its first comment's author
|
|
authors = [((t["comments"]["nodes"] or [{}])[0].get("author") or {}).get("login") or ""
|
|
for t in threads["nodes"] if not t["isResolved"]]
|
|
state.update(number=nodes[0]["number"], open=len(authors),
|
|
codex=sum(1 for a in authors if PR_CODEX.search(a)),
|
|
more=threads["pageInfo"]["hasNextPage"])
|
|
except Exception:
|
|
state["error"] = True
|
|
try:
|
|
os.makedirs(TOKENS_STATE_DIR, exist_ok=True)
|
|
tmp = f"{path}.{os.getpid()}.tmp"
|
|
with open(tmp, "w") as fh:
|
|
json.dump(state, fh)
|
|
os.replace(tmp, path)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def pr_segment(cwd, top):
|
|
"""Unresolved review threads (any author, outdated ones included) on the current branch's
|
|
open PR: `💬 #N ✓` when none are left, else the count with Codex's share broken out.
|
|
|
|
A GitHub round trip takes ~0.7 s, too slow for a render, so this only reads a per-branch
|
|
cache (~/.claude/cache/statusline.pr.<hash>.json) and, once it is older than PR_TTL,
|
|
spawns a detached `--pr-refresh` child to update it -- the new count shows on the next
|
|
render. PR-conversation comments are not counted: they have no resolved state, so a
|
|
summary or an LGTM would keep the count from ever reaching zero.
|
|
"""
|
|
try:
|
|
if not top or not shutil.which("gh"):
|
|
return ""
|
|
branch = _git(cwd, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
|
|
if not branch or branch == "HEAD":
|
|
return ""
|
|
path = _pr_cache_path(top, branch)
|
|
state = _read_json(path)
|
|
if time.time() - (state.get("ts") or 0) > (PR_ERR_TTL if state.get("error") else PR_TTL):
|
|
claim = path + ".lock"
|
|
try:
|
|
claimed = time.time() - os.path.getmtime(claim) < PR_CLAIM
|
|
except OSError:
|
|
claimed = False
|
|
if not claimed:
|
|
os.makedirs(TOKENS_STATE_DIR, exist_ok=True)
|
|
with open(claim, "w"):
|
|
pass
|
|
# own session + no inherited stdout: Claude Code reads ours to EOF, and
|
|
# cancels an in-flight render when the next one triggers
|
|
subprocess.Popen(
|
|
[sys.executable, os.path.abspath(__file__), "--pr-refresh", cwd, branch, path],
|
|
stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
start_new_session=True,
|
|
)
|
|
num = state.get("number")
|
|
if not num:
|
|
return ""
|
|
n, codex = state.get("open") or 0, state.get("codex") or 0
|
|
out = "💬 " + color(f"#{num} ", "gray")
|
|
if not n:
|
|
out += color("✓", "green")
|
|
else:
|
|
out += color(f"{n}{'+' if state.get('more') else ''}", "yellow", bold=True)
|
|
if codex:
|
|
out += color(" · ", "gray") + color(f"codex {codex}", "red")
|
|
if state.get("error"):
|
|
out += color(" ?", "gray") # last fetch failed: these counts are stale
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def model_segment(data):
|
|
name = (data.get("model") or {}).get("display_name")
|
|
if not name:
|
|
return ""
|
|
name = name.replace(" context)", ")") # "Opus 4.8 (1M context)" -> "Opus 4.8 (1M)"
|
|
out = "🤖 " + color(name, "blue")
|
|
level = (data.get("effort") or {}).get("level")
|
|
if level:
|
|
out += color(f" · {level}", "gray")
|
|
return out
|
|
|
|
|
|
def context_window(data):
|
|
"""Size of this session's context window, in tokens.
|
|
|
|
Claude Code resolves the `[1m]` model suffix away before handing the status line
|
|
its JSON -- `model.id` is always the bare API id ("claude-fable-5") -- so sniffing
|
|
the id for "1m" could never detect a 1M session and every one of them was measured
|
|
against a 200k window (5x overstated %, ⚠compact from ~16% real usage). The payload
|
|
carries the real number; display_name ("Opus 4.8 (1M context)") is the fallback.
|
|
"""
|
|
cw = data.get("context_window")
|
|
if isinstance(cw, dict):
|
|
for key in ("context_window_size", "max_tokens", "size"):
|
|
val = cw.get(key)
|
|
if isinstance(val, (int, float)) and val > 0:
|
|
return int(val)
|
|
name = ((data.get("model") or {}).get("display_name") or "").lower()
|
|
return 1_000_000 if "1m" in name else 200_000
|
|
|
|
|
|
def context_segment(data, usage):
|
|
try:
|
|
if not usage:
|
|
return ""
|
|
used = (
|
|
usage.get("input_tokens", 0)
|
|
+ usage.get("cache_creation_input_tokens", 0)
|
|
+ usage.get("cache_read_input_tokens", 0)
|
|
)
|
|
window = context_window(data)
|
|
pct = used / window * 100 if window else 0
|
|
# Prefer Claude Code's own percentage when the payload carries it (2.1.x) -- it
|
|
# already accounts for the auto-compact reserve and window quirks we can't see.
|
|
cw = data.get("context_window")
|
|
if isinstance(cw, dict) and isinstance(cw.get("used_percentage"), (int, float)):
|
|
pct = float(cw["used_percentage"])
|
|
col = bucket(pct, 50, 80)
|
|
out = "📝 " + color(f"{pct:.0f}%", col) + color(f" ({used / 1000:.0f}k)", "gray", dim=True)
|
|
if pct >= 80:
|
|
out += color(" ⚠compact", "red", bold=True)
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def usage_segment(data):
|
|
"""5-hour rolling usage window: used % (+ time until it resets)."""
|
|
try:
|
|
five = (data.get("rate_limits") or {}).get("five_hour") or {}
|
|
pct = five.get("used_percentage")
|
|
if pct is None:
|
|
return ""
|
|
out = "📊 " + color(f"{pct:.0f}%", bucket(pct, 50, 80))
|
|
resets = five.get("resets_at")
|
|
if resets:
|
|
secs = int(resets) - int(time.time())
|
|
if secs > 0:
|
|
h, m = divmod(secs // 60, 60)
|
|
out += color(f" {h}h{m:02d}m" if h else f" {m}m", "gray")
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def cost_segment(data):
|
|
try:
|
|
cost = data.get("cost") or {}
|
|
usd = cost.get("total_cost_usd")
|
|
if usd is None:
|
|
return ""
|
|
out = "💰 " + color(f"${usd:.2f}", "yellow")
|
|
dur_ms = cost.get("total_duration_ms") or 0
|
|
if dur_ms > 30_000: # need a meaningful window before extrapolating
|
|
per_hr = usd / (dur_ms / 3_600_000)
|
|
out += color(f" ${per_hr:.2f}/h", "gray", dim=True)
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def velocity_segment(data):
|
|
try:
|
|
cost = data.get("cost") or {}
|
|
added = cost.get("total_lines_added", 0)
|
|
removed = cost.get("total_lines_removed", 0)
|
|
if not (added or removed):
|
|
return ""
|
|
out = "✏️ " + color(f"+{added}", "green") + color("/", "gray", dim=True) + color(f"-{removed}", "red")
|
|
dur_ms = cost.get("total_duration_ms") or 0
|
|
if dur_ms > 30_000:
|
|
per_min = added / (dur_ms / 60_000)
|
|
out += color(f" {per_min:.0f}/m", "gray", dim=True)
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def tokens_segment(totals):
|
|
"""Cumulative session tokens: ↑ input (incl. cache reads/writes) and ↓ output."""
|
|
try:
|
|
if not totals:
|
|
return ""
|
|
tin, tout = totals["in"], totals["out"]
|
|
if not (tin or tout):
|
|
return ""
|
|
return "🔢 " + color(f"↑{fmt_tokens(tin)}", "cyan") + " " + color(f"↓{fmt_tokens(tout)}", "green")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def tps_segment(totals):
|
|
"""Decode speed in tokens/s: duration-weighted mean over the last few thinking blocks
|
|
(see session_tokens). Omitted until a thinking block ≥1s has completed -- so never
|
|
shown with thinking disabled."""
|
|
try:
|
|
samples = (totals or {}).get("tps") or []
|
|
toks = sum(s[0] for s in samples)
|
|
ms = sum(s[1] for s in samples)
|
|
if toks <= 0 or ms <= 0:
|
|
return ""
|
|
rate = toks / ms * 1000
|
|
return "⚡ " + color(f"{rate:.0f} t/s", bucket(rate, 20, 40, invert=True))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def cache_segment(usage, data=None):
|
|
"""Prompt-cache hit % (higher is better) + the cache TTL, `·cold` when the cache has
|
|
expired. Hit % from the 2.1.260+ payload `prompt_cache.hit_ratio` (a 0..1 fraction, null
|
|
while all counts are zero), else computed from the usage block; ttl/cold always from the
|
|
payload when present. A 1h TTL is what keeps a 10-minute pr-loop cadence warm."""
|
|
try:
|
|
pc = (data or {}).get("prompt_cache")
|
|
pc = pc if isinstance(pc, dict) else {}
|
|
pct = None
|
|
if isinstance(pc.get("hit_ratio"), (int, float)):
|
|
pct = float(pc["hit_ratio"]) * 100
|
|
elif usage:
|
|
read = usage.get("cache_read_input_tokens", 0)
|
|
total = usage.get("input_tokens", 0) + usage.get("cache_creation_input_tokens", 0) + read
|
|
if total > 0:
|
|
pct = read / total * 100
|
|
if pct is None:
|
|
return ""
|
|
out = "♻️ " + color(f"{pct:.0f}%", bucket(pct, 50, 80, invert=True))
|
|
ttl = pc.get("ttl")
|
|
if isinstance(ttl, str) and ttl:
|
|
out += color(f" {ttl}", "gray")
|
|
if pc.get("warm") is False:
|
|
out += color(" ·cold", "yellow")
|
|
return out
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def api_segment(data):
|
|
"""Share of wall-clock time spent in API/inference (rough 'busy' ratio)."""
|
|
try:
|
|
cost = data.get("cost") or {}
|
|
wall = cost.get("total_duration_ms") or 0
|
|
api = cost.get("total_api_duration_ms") or 0
|
|
if wall <= 0 or api <= 0:
|
|
return ""
|
|
pct = min(api / wall * 100, 100)
|
|
return "⚙️ " + color(f"{pct:.0f}%", "blue")
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
# --- row 2: system ----------------------------------------------------------
|
|
def ram_segment():
|
|
try:
|
|
info = {}
|
|
with open("/proc/meminfo") as fh:
|
|
for line in fh:
|
|
k, _, v = line.partition(":")
|
|
info[k.strip()] = int(v.split()[0]) # kB
|
|
total = info.get("MemTotal", 0)
|
|
avail = info.get("MemAvailable", info.get("MemFree", 0))
|
|
if total <= 0:
|
|
return ""
|
|
used = total - avail
|
|
pct = used / total * 100
|
|
g = 1024 * 1024
|
|
return (
|
|
"🧠 " + color(f"{pct:.0f}%", bucket(pct, 70, 85))
|
|
+ color(f" {used / g:.1f}/{total / g:.0f}G", "gray", dim=True)
|
|
)
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def disk_segment(cwd):
|
|
try:
|
|
st = os.statvfs(cwd)
|
|
total = st.f_blocks
|
|
if total <= 0:
|
|
return ""
|
|
used = total - st.f_bfree
|
|
pct = used / total * 100
|
|
return "💾 " + color(f"{pct:.0f}%", bucket(pct, 75, 90))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def _cpu_pct():
|
|
"""Live CPU% diffed against a cached /proc/stat snapshot (no sleeping)."""
|
|
try:
|
|
with open("/proc/stat") as fh:
|
|
parts = fh.readline().split()
|
|
if not parts or parts[0] != "cpu":
|
|
return None
|
|
vals = [int(x) for x in parts[1:]]
|
|
idle = vals[3] + (vals[4] if len(vals) > 4 else 0) # idle + iowait
|
|
total = sum(vals)
|
|
|
|
state = os.path.expanduser("~/.claude/cache/statusline.cpu")
|
|
prev = None
|
|
try:
|
|
with open(state) as fh:
|
|
pt, pi = fh.read().split()
|
|
prev = (int(pt), int(pi))
|
|
except Exception:
|
|
prev = None
|
|
try:
|
|
os.makedirs(os.path.dirname(state), exist_ok=True)
|
|
with open(state, "w") as fh:
|
|
fh.write(f"{total} {idle}")
|
|
except Exception:
|
|
pass
|
|
|
|
if not prev:
|
|
return None
|
|
dt = total - prev[0]
|
|
di = idle - prev[1]
|
|
if dt <= 0 or di < 0:
|
|
return None
|
|
return max(0.0, min(100.0, (1 - di / dt) * 100))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def cpu_segment():
|
|
try:
|
|
load1 = os.getloadavg()[0]
|
|
cores = os.cpu_count() or 1
|
|
pct = _cpu_pct()
|
|
if pct is not None:
|
|
head = color(f"{pct:.0f}%", bucket(pct, 60, 85))
|
|
tail = color(f" {load1:.1f} {cores}c", "gray", dim=True)
|
|
else:
|
|
head = color(f"{load1:.2f}", bucket(load1 / cores, 0.7, 1.0))
|
|
tail = color(f" {cores}c", "gray", dim=True)
|
|
return "🖥️ " + head + tail
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def temp_segment():
|
|
"""Hottest CPU-ish thermal zone in °C, falling back to the max zone."""
|
|
try:
|
|
base = "/sys/class/thermal"
|
|
if not os.path.isdir(base):
|
|
return ""
|
|
prefer = ("x86_pkg_temp", "coretemp", "cpu", "k10temp", "tctl")
|
|
chosen = None
|
|
fallback = None
|
|
for name in os.listdir(base):
|
|
if not name.startswith("thermal_zone"):
|
|
continue
|
|
d = os.path.join(base, name)
|
|
try:
|
|
with open(os.path.join(d, "temp")) as fh:
|
|
milli = int(fh.read().strip())
|
|
except Exception:
|
|
continue
|
|
ztype = ""
|
|
try:
|
|
with open(os.path.join(d, "type")) as fh:
|
|
ztype = fh.read().strip().lower()
|
|
except Exception:
|
|
pass
|
|
if fallback is None or milli > fallback:
|
|
fallback = milli
|
|
if any(p in ztype for p in prefer) and (chosen is None or milli > chosen):
|
|
chosen = milli
|
|
milli = chosen if chosen is not None else fallback
|
|
if milli is None:
|
|
return ""
|
|
c = milli / 1000.0
|
|
return "🌡️ " + color(f"{c:.0f}°C", bucket(c, 60, 80))
|
|
except Exception:
|
|
return ""
|
|
|
|
|
|
def version_segment(data):
|
|
v = data.get("version")
|
|
return color(f"v{v}", "gray", dim=True) if v else ""
|
|
|
|
|
|
def style_segment(data):
|
|
name = (data.get("output_style") or {}).get("name")
|
|
return color(f"🎨{name}", "gray", dim=True) if name else ""
|
|
|
|
|
|
def main():
|
|
if sys.argv[1:2] == ["--pr-refresh"] and len(sys.argv) == 5:
|
|
pr_refresh(*sys.argv[2:])
|
|
return
|
|
try:
|
|
data = json.load(sys.stdin)
|
|
except Exception:
|
|
data = {}
|
|
|
|
cwd = (
|
|
(data.get("workspace") or {}).get("current_dir")
|
|
or data.get("cwd")
|
|
or os.getcwd()
|
|
)
|
|
usage = payload_usage(data) or last_usage(data)
|
|
totals = session_tokens(data)
|
|
top = git_top(cwd)
|
|
|
|
# One row: work (left) then system (right), joined. If it would overflow the
|
|
# terminal, trailing (lowest-priority) segments are dropped so it stays one line.
|
|
# Dormant helpers kept above for easy re-add: ram_segment, cpu_segment,
|
|
# temp_segment (retired -- they only refresh on message events, so they sat
|
|
# stale; the tmux status bar owns system metrics now), velocity_segment,
|
|
# api_segment, version_segment, style_segment.
|
|
work = [s for s in (
|
|
vim_segment(data),
|
|
dir_segment(cwd, top),
|
|
git_segment(cwd, top),
|
|
pr_segment(cwd, top),
|
|
model_segment(data),
|
|
context_segment(data, usage),
|
|
usage_segment(data),
|
|
cost_segment(data),
|
|
tokens_segment(totals),
|
|
tps_segment(totals),
|
|
cache_segment(usage, data),
|
|
) if s]
|
|
system = [s for s in (
|
|
disk_segment(cwd),
|
|
) if s]
|
|
|
|
segs = work + system
|
|
cols = term_cols()
|
|
if cols: # trim from the right until it fits one line (cols=0 -> width unknown, keep all)
|
|
while len(segs) > 1 and vlen(join_line(segs)) > cols:
|
|
segs.pop()
|
|
sys.stdout.write(join_line(segs))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|