状态栏不能卡：把 Git、耗时和缓存都塞进三行
Claude Code 的状态栏脚本，我写过好几版。前几版都在加信息：加 Git 分支，加代码行数，加 token 速度。加到最后，我自己都担心一件事——它会不会把终端拖慢。
状态栏是每次刷新都要跑的。它慢 100 毫秒，你感觉不到；慢 2 秒，你就开始怀疑 Claude Code 卡了。所以这一版我给自己定了个规矩：主路径只做四件事，解析 stdin、跑一次 git、读一次 transcript 尾部、拼字符串。不读设置文件，不发网络请求。
完整代码
全文件四百多行，先贴出来，下面只讲几个决定。
python
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Claude Code 状态栏：读 stdin JSON，渲染三行。
第一行：模型 / 上下文窗口。
第二行：目录 / Git 状态 / 代码量。
第三行：TTFT / TPOT / TPS / 缓存命中。
主路径只做四件事：解析 stdin、跑一次 git、读一次 transcript 尾部、拼字符串，
不读设置文件、不发网络请求。
"""
import json
import os
import subprocess
import sys
from datetime import datetime
GIT_TIMEOUT_SECONDS = 0.15
# 最近一次请求的 api-request 行和回复总在 transcript 末尾附近，只读尾部就够了。
TRANSCRIPT_TAIL_BYTES = 256 * 1024
CHANGED_STATUS_TAGS = ("1 ", "2 ")
NAMED_HEADS = {"(detached)", "(unknown)"}
# ------------------------------------------------------------
# 通用工具
# ------------------------------------------------------------
def as_mapping(value: object) -> dict:
return value if isinstance(value, dict) else {}
def as_int(value: object, default: int = 0) -> int:
try:
return int(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return default
def shorten(text: str, limit: int = 32) -> str:
return text if len(text) <= limit else f"{text[: limit - 1]}…"
def join_segments(*segments: str) -> str:
return " | ".join(segment for segment in segments if segment)
# ------------------------------------------------------------
# 输入
# ------------------------------------------------------------
def read_input() -> dict:
try:
data = json.loads(sys.stdin.buffer.read())
except (OSError, ValueError):
return {}
return data if isinstance(data, dict) else {}
# ------------------------------------------------------------
# 第一行：模型 / 上下文窗口
# ------------------------------------------------------------
def model_line(data: dict) -> str:
return join_segments(model_segment(data), context_segment(data))
def model_segment(data: dict) -> str:
model = as_mapping(data.get("model"))
effort = as_mapping(data.get("effort")).get("level") or "-"
return f"model: {model.get('id') or 'unknown'}({effort})"
def context_segment(data: dict) -> str:
window = as_mapping(data.get("context_window"))
size = format_token_count(window.get("context_window_size"))
used = as_int(window.get("used_percentage"))
return f"window[{size}]: {used}%"
def format_token_count(value: object) -> str:
count = as_int(value)
if count >= 1_000_000:
return f"{count // 1_000_000}M"
if count >= 1_000:
return f"{count // 1_000}K"
return str(count)
# ------------------------------------------------------------
# 第二行：目录 / Git 状态 / 代码量
# ------------------------------------------------------------
def workspace_line(data: dict) -> str:
cwd = current_dir(data)
return join_segments(dir_segment(cwd), git_segment(cwd), code_segment(data))
def current_dir(data: dict) -> str:
workspace = as_mapping(data.get("workspace"))
return (
workspace.get("current_dir")
or data.get("cwd")
or workspace.get("project_dir")
or ""
)
def dir_segment(cwd: str) -> str:
name = os.path.basename(cwd.rstrip("/\\"))
return f"dir: {name}" if name else ""
def code_segment(data: dict) -> str:
cost = as_mapping(data.get("cost"))
added = as_int(cost.get("total_lines_added"))
removed = as_int(cost.get("total_lines_removed"))
return f"code:+{added}/-{removed}"
# ------------------------------------------------------------
# Transcript 读取
# ------------------------------------------------------------
def read_transcript_tail(path: object) -> list[bytes]:
if not isinstance(path, str) or not path:
return []
try:
with open(path, "rb") as file:
file.seek(0, os.SEEK_END)
size = file.tell()
file.seek(max(size - TRANSCRIPT_TAIL_BYTES, 0))
raw = file.read()
except OSError:
return []
if size > TRANSCRIPT_TAIL_BYTES:
_, _, raw = raw.partition(b"\n") # 首行多半被截断
return raw.splitlines()
def parse_record(line: bytes) -> dict | None:
try:
record = json.loads(line)
except ValueError:
return None
return record if isinstance(record, dict) else None
def parse_timestamp(value: object) -> datetime | None:
if not isinstance(value, str):
return None
try:
return datetime.fromisoformat(value)
except ValueError:
return None
# ------------------------------------------------------------
# 第三行：TTFT / TPOT / TPS / 缓存命中
# ------------------------------------------------------------
class RequestSpeed:
"""一次请求的耗时拆分。时间戳都来自 transcript 落盘时刻，不是精确的流式时间。"""
__slots__ = ("output_tokens", "total_seconds", "ttft_seconds")
def __init__(
self, ttft_seconds: float, total_seconds: float, output_tokens: int
) -> None:
self.ttft_seconds = ttft_seconds
self.total_seconds = total_seconds
self.output_tokens = output_tokens
def speed_line(data: dict) -> str:
speed = latest_request_speed(data)
return join_segments(
ttft_segment(speed),
tpot_segment(speed),
tps_segment(speed),
cache_segment(data),
)
def latest_request_speed(data: dict) -> RequestSpeed | None:
"""从 transcript 尾部倒扫最近一次有回复的主线程请求。
状态栏输入不带耗时字段，只能估算：
- TTFT = 首个内容块落盘时间 - 请求发出时间（含 thinking 耗时）
- TPS = 输出 token 数 / 整段请求耗时
"""
replies: list[dict] = []
for line in reversed(read_transcript_tail(data.get("transcript_path"))):
record = parse_record(line)
if record is None or record.get("isSidechain"):
continue
if record.get("type") == "assistant":
replies.append(record)
elif record.get("type") == "api-request":
speed = request_speed(record, replies)
if speed:
return speed
return None
def request_speed(request: dict, replies: list[dict]) -> RequestSpeed | None:
started_at = parse_timestamp(request.get("timestamp"))
matched = [
reply for reply in replies if reply.get("requestRef") == request.get("id")
]
reply_times = [
moment
for moment in (parse_timestamp(reply.get("timestamp")) for reply in matched)
if moment
]
tokens = max((reply_output_tokens(reply) for reply in matched), default=0)
if started_at is None or not reply_times or tokens <= 0:
return None
first_at, last_at = min(reply_times), max(reply_times)
return RequestSpeed(
(first_at - started_at).total_seconds(),
(last_at - started_at).total_seconds(),
tokens,
)
def reply_output_tokens(reply: dict) -> int:
usage = as_mapping(as_mapping(reply.get("message")).get("usage"))
return as_int(usage.get("output_tokens"))
def format_seconds(seconds: float) -> str:
return f"{seconds:.1f}s" if seconds >= 1 else f"{seconds * 1000:.0f}ms"
def ttft_segment(speed: RequestSpeed | None) -> str:
if not speed:
return "ttft: -"
return f"ttft: {format_seconds(speed.ttft_seconds)}"
def tpot_segment(speed: RequestSpeed | None) -> str:
if not speed or speed.total_seconds <= 0:
return "tpot: -"
return f"tpot: {speed.total_seconds / speed.output_tokens * 1000:.1f}ms"
def tps_segment(speed: RequestSpeed | None) -> str:
if not speed or speed.total_seconds <= 0:
return "tps: -"
return f"tps: {speed.output_tokens / speed.total_seconds:.0f}"
def cache_segment(data: dict) -> str:
ratio = as_mapping(data.get("prompt_cache")).get("hit_ratio")
if not isinstance(ratio, (int, float)):
percent = 0
elif ratio <= 1: # 接口可能给 0~1 的小数，也可能直接给百分数
percent = round(ratio * 100)
else:
percent = round(ratio)
return f"cache: {percent}%"
# ------------------------------------------------------------
# Git 状态
# ------------------------------------------------------------
class GitStatus:
"""git status --porcelain=v2 的解析结果。"""
__slots__ = (
"ahead",
"behind",
"conflicts",
"deleted",
"has_ahead_behind",
"head",
"modified",
"oid",
"staged",
"untracked",
"upstream",
)
def __init__(self) -> None:
self.head = self.oid = self.upstream = ""
self.has_ahead_behind = False
self.ahead = self.behind = 0
self.conflicts = self.staged = self.modified = self.deleted = 0
self.untracked = 0
def git_segment(cwd: str) -> str:
if not cwd:
return ""
status = run_git_status(cwd)
if not status:
return ""
state = parse_git_status(status)
parts = [f"git:{git_ref(state)}", *sync_parts(state), *change_parts(state)]
return join_segments(*parts)
def run_git_status(cwd: str) -> str:
try:
result = subprocess.run(
[
"git",
"-C",
cwd,
"status",
"--porcelain=v2",
"--branch",
"--untracked-files=normal",
],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
timeout=GIT_TIMEOUT_SECONDS,
check=False,
)
except (OSError, subprocess.SubprocessError, UnicodeDecodeError):
return ""
return result.stdout if result.returncode == 0 else ""
def parse_git_status(status: str) -> GitStatus:
state = GitStatus()
for line in status.splitlines():
if line.startswith("# branch."):
parse_branch_header(line, state)
elif line.startswith("? "):
state.untracked += 1
elif line.startswith("u "):
state.conflicts += 1
elif line.startswith(CHANGED_STATUS_TAGS):
count_change(line, state)
return state
def parse_branch_header(line: str, state: GitStatus) -> None:
key, _, value = line.removeprefix("# branch.").partition(" ")
if key == "head":
state.head = value
elif key == "oid":
state.oid = value
elif key == "upstream":
state.upstream = value
elif key == "ab":
state.has_ahead_behind = True
for flag in value.split():
if flag.startswith("+"):
state.ahead = as_int(flag[1:])
elif flag.startswith("-"):
state.behind = as_int(flag[1:])
def count_change(line: str, state: GitStatus) -> None:
fields = line.split(maxsplit=2)
if len(fields) < 2:
return
index_status, worktree_status = fields[1][:2]
if "D" in (index_status, worktree_status):
state.deleted += 1
if index_status not in ".D":
state.staged += 1
if worktree_status not in ".D":
state.modified += 1
def git_ref(state: GitStatus) -> str:
if state.head and state.head not in NAMED_HEADS:
return shorten(state.head)
if state.oid and state.oid != "(initial)":
return f"@{state.oid[:7]}"
return "unknown"
def sync_parts(state: GitStatus) -> list[str]:
named_head = bool(state.head) and state.head not in NAMED_HEADS
if not state.upstream:
return ["no-up"] if named_head else []
if not state.has_ahead_behind:
return ["no-ab"]
parts = []
if state.ahead:
parts.append(f"↑{state.ahead}")
if state.behind:
parts.append(f"↓{state.behind}")
return parts or ["sync"]
def change_parts(state: GitStatus) -> list[str]:
counts = (
(state.conflicts, "!"),
(state.staged, "+"),
(state.modified, "~"),
(state.deleted, "-"),
(state.untracked, "?"),
)
parts = [f"{token}{count}" for count, token in counts if count]
return parts or ["clean"]
# ------------------------------------------------------------
# 渲染与入口
# ------------------------------------------------------------
def render(data: dict) -> str:
lines = (
model_line(data),
workspace_line(data),
speed_line(data),
)
return "\n".join(line for line in lines if line)
def main() -> int:
print(render(read_input()))
return 0
if __name__ == "__main__":
raise SystemExit(main())三行各自管什么
第一行是模型和上下文窗口：
txt
model: claude-opus-5-5(max) | window[1M]: 23%第二行是目录、Git 状态和代码量：
txt
dir: nine-home | git:main ↑2 +3 ~1 ?1 | code:+184/-96第三行是这次请求的速度：
txt
ttft: 860ms | tpot: 18.2ms | tps: 55 | cache: 96%信息密度不低，但每一段都短，扫一眼就知道有没有异常。
Git 只跑一次，超时就放弃
Git 状态是整条链路里最可能慢的部分。仓库大、磁盘慢、.git 在 WSL 的 /mnt/c 上，随便哪个都能让 git status 拖到几百毫秒。
我的做法是：
python
def run_git_status(cwd: str) -> str:
try:
result = subprocess.run(
[
"git", "-C", cwd, "status",
"--porcelain=v2", "--branch", "--untracked-files=normal",
],
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
timeout=GIT_TIMEOUT_SECONDS,
check=False,
)
except (OSError, subprocess.SubprocessError, UnicodeDecodeError):
return ""
return result.stdout if result.returncode == 0 else ""GIT_TIMEOUT_SECONDS = 0.15。超时就返回空字符串，状态栏里直接不显示 Git 段。
既不显示 git: error，也不弹通知。这一格直接空着。Git 偶尔慢一次，我不想被它打断。
另一个细节是 --porcelain=v2 --branch。一次调用就能拿到分支名、上游、领先落后数量和所有文件状态，不用再跑 rev-parse 和 status --short。状态栏这种高频脚本，少一次子进程就是省一次开销。
解析部分也全用纯函数，parse_git_status 收字符串、吐 GitStatus。出问题的时候，我把一段 git status 输出丢进去就能复现，不用真的去弄一个坏仓库。
耗时从 transcript 的尾巴里估
状态栏的输入 JSON 不带耗时字段。TTFT、TPS 这些只能自己算。
Claude Code 会把 transcript 写到磁盘，每次请求有一条 api-request 记录，回复有 assistant 记录。我在文件尾部往回扫，找到最近一次有回复的主线程请求：
python
def latest_request_speed(data: dict) -> RequestSpeed | None:
replies: list[dict] = []
for line in reversed(read_transcript_tail(data.get("transcript_path"))):
record = parse_record(line)
if record is None or record.get("isSidechain"):
continue
if record.get("type") == "assistant":
replies.append(record)
elif record.get("type") == "api-request":
speed = request_speed(record, replies)
if speed:
return speed
return None关键是 read_transcript_tail：
python
TRANSCRIPT_TAIL_BYTES = 256 * 1024
def read_transcript_tail(path: object) -> list[bytes]:
...
file.seek(0, os.SEEK_END)
size = file.tell()
file.seek(max(size - TRANSCRIPT_TAIL_BYTES, 0))
raw = file.read()
...
if size > TRANSCRIPT_TAIL_BYTES:
_, _, raw = raw.partition(b"\n") # 首行多半被截断transcript 可能几十兆，我不读它。只读最后 256KB，第一行被截断就丢掉。最近一次请求一定在这段里，除非这次请求特别长——那也没关系，显示 - 而已。
这里还做了一件事：跳过 isSidechain。子代理的请求不该出现在主线程的状态栏里，否则你看到的 TPS 可能是某个后台任务的速度。
时间口径写清楚
TTFT 的算法是「首个回复落盘时间减去请求发出时间」。这包含了 thinking 的时间，不是严格的「第一个 token 到达」。所以我在类注释里写清楚了：
python
class RequestSpeed:
"""一次请求的耗时拆分。时间戳都来自 transcript 落盘时刻，不是精确的流式时间。"""TPOT 用总耗时除以输出 token 数，TPS 反过来。都不是实验室级别的精度，但足够回答「这次请求快不快」。
缓存比例兼容两种口径
接口有时候给 0.96，有时候直接给 96。这版这样处理：
python
def cache_segment(data: dict) -> str:
ratio = as_mapping(data.get("prompt_cache")).get("hit_ratio")
if not isinstance(ratio, (int, float)):
percent = 0
elif ratio <= 1: # 接口可能给 0~1 的小数，也可能直接给百分数
percent = round(ratio * 100)
else:
percent = round(ratio)
return f"cache: {percent}%"一行判断，两种都吃得下。边界情况也想过了：1 按小数算，是 100%，正好对；0 是 0%，也对。
写状态栏和写库不是一回事
这脚本没有测试，没有抽象层，没有插件机制。它就是一条从 stdin 到 stdout 的直线，中间分叉成几个小函数，每个函数都短到可以一眼看完。
但它有几个明确的取舍：
- 超时就空着，不报错
- 读尾部，不读全文
- 拿不到就显示
-，不猜 - 所有解析都是纯函数，输入输出清楚
切目录、跑测试、改文件的时候，它都在下面安静地换数字。没有一次让我等过。