[Update] bulk update
This commit is contained in:
@@ -90,25 +90,49 @@ def term_cols():
|
||||
|
||||
|
||||
# --- shared transcript usage ------------------------------------------------
|
||||
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 {}."""
|
||||
"""Most recent `message.usage` block from the transcript, or {}.
|
||||
|
||||
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 {}
|
||||
usage = {}
|
||||
with open(path, "r", errors="ignore") as fh:
|
||||
for line in fh:
|
||||
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"]
|
||||
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 {}
|
||||
@@ -193,7 +217,14 @@ def git_segment(cwd, top):
|
||||
return ""
|
||||
repo = os.path.basename(top) or "repo"
|
||||
branch = _git(cwd, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip() or "?"
|
||||
dirty = bool(_git(cwd, "status", "--porcelain").stdout.strip())
|
||||
|
||||
# `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 = ""
|
||||
@@ -231,6 +262,25 @@ def model_segment(data):
|
||||
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:
|
||||
@@ -240,8 +290,7 @@ def context_segment(data, usage):
|
||||
+ usage.get("cache_creation_input_tokens", 0)
|
||||
+ usage.get("cache_read_input_tokens", 0)
|
||||
)
|
||||
model_id = (data.get("model") or {}).get("id", "")
|
||||
window = 1_000_000 if "1m" in model_id.lower() else 200_000
|
||||
window = context_window(data)
|
||||
pct = used / window * 100 if window else 0
|
||||
col = bucket(pct, 50, 80)
|
||||
out = "📝 " + color(f"{pct:.0f}%", col) + color(f" ({used / 1000:.0f}k)", "gray", dim=True)
|
||||
|
||||
Reference in New Issue
Block a user