Claude Code status line — two-line layout with context, quota, and cost bars

statusline-command.py

#!/usr/bin/env python3
"""Claude Code status line — two-row, three-column box.

Row 1: Model/Dir/Duration  │  Cost (burn rate)  │  Token velocity
Row 2: Context bar (used/tot)  │  5h quota bar ⟳reset  │  7d quota bar ⟳reset

Rendered inside a box-drawing border (┌─┬─┐ │ └─┴─┘) with centered cells,
color-coded progress bars, and hot pink pacing markers.
"""

import json
import re
import subprocess
import sys
import time
import urllib.request
from datetime import datetime
from pathlib import Path

# ---------------------------------------------------------------------------
# ANSI helpers
# ---------------------------------------------------------------------------

RESET = "\033[0m"
DIM = "\033[2m"
GRAY = "\033[38;5;242m"
WHITE = "\033[38;5;255m"
CYAN = "\033[96m"
HOT_PINK = "\033[38;5;199m"

# Subdued bar colors (non-bold, 256-color)
BAR_GREEN = "\033[38;5;65m"
BAR_YELLOW = "\033[38;5;137m"
BAR_RED = "\033[38;5;131m"

ANSI_RE = re.compile(r"\033\[[0-9;]*m")


def dim(s: str) -> str:
    return f"{DIM}{s}{RESET}"


def visible_len(s: str) -> int:
    """Length of a string after stripping ANSI escape codes."""
    return len(ANSI_RE.sub("", s))


def render_grid(rows: list[list[str]]) -> str:
    """Render rows inside a box-drawing border with centered cells.

    Each row is a list of cell strings (may contain ANSI codes).
    Columns are padded to the max visible width across all rows,
    then enclosed in a ┌─┬─┐ / │ │ │ / └─┴─┘ box.
    """
    if not rows:
        return ""

    num_cols = max(len(r) for r in rows)

    # Find max visible width for each column
    col_widths = [0] * num_cols
    for row in rows:
        for c, cell in enumerate(row):
            col_widths[c] = max(col_widths[c], visible_len(cell))

    lines: list[str] = []
    pipe = f"{DIM}{GRAY}│{RESET}"
    sep = f" {pipe} "

    def rule(left_corner: str, mid: str, right_corner: str) -> str:
        parts = []
        for c in range(num_cols):
            parts.append("─" * (col_widths[c] + 2))
            if c < num_cols - 1:
                parts.append(mid)
        return f"{DIM}{GRAY}{left_corner}{''.join(parts)}{right_corner}{RESET}"

    lines.append(rule("┌", "┬", "┐"))

    for row in rows:
        parts: list[str] = []
        for c in range(num_cols):
            cell = row[c] if c < len(row) else ""
            pad = col_widths[c] - visible_len(cell)
            left = pad // 2
            right = pad - left
            parts.append(" " * left + cell + " " * right)
        lines.append(f"{pipe} " + sep.join(parts) + f" {pipe}")

    lines.append(rule("└", "┴", "┘"))

    return "\n".join(lines)


def pct_color(
    pct: float, green_thresh: int = 50, yellow_thresh: int = 80,
) -> str:
    """Return ANSI color code for a usage percentage."""
    p = int(round(pct))
    if p < green_thresh:
        return BAR_GREEN
    elif p < yellow_thresh:
        return BAR_YELLOW
    else:
        return BAR_RED


# ---------------------------------------------------------------------------
# Formatting helpers
# ---------------------------------------------------------------------------

def format_k(val: int) -> str:
    return f"{val // 1000}k" if val >= 1000 else str(val)


def format_tok(val: int) -> str:
    sign = "+" if val >= 0 else ""
    if abs(val) >= 1000:
        return f"{sign}{val / 1000:.1f}k"
    return f"{sign}{val}"


def format_ema(val: float) -> str:
    if abs(val) >= 1000:
        return f"{val / 1000:.1f}k"
    return f"{int(round(val))}"


def format_duration(ms: float) -> str:
    s = int(ms) // 1000
    h, s = divmod(s, 3600)
    m, s = divmod(s, 60)
    if h > 0:
        return f"{h}h{m}m"
    if m > 0:
        return f"{m}m{s}s"
    return f"{s}s"


# ---------------------------------------------------------------------------
# Progress bar
# ---------------------------------------------------------------------------

def build_bar(
    pct: float,
    width: int = 20,
    target_pct: float | None = None,
    green_thresh: int = 50,
    yellow_thresh: int = 80,
) -> str:
    """Build a colored progress bar string.

    If target_pct is given, a hot pink │ pacing marker is placed at that
    position in the bar.
    """
    used_int = int(round(pct))
    filled = min(used_int * width // 100, width)
    bar_color = pct_color(pct, green_thresh, yellow_thresh)

    target_pos = -1
    if target_pct is not None:
        target_pos = int(round(target_pct)) * width // 100
        target_pos = max(0, min(target_pos, width - 1))

    # Build in contiguous color segments to minimize ANSI codes.
    segments: list[str] = []
    i = 0
    while i < width:
        if i == target_pos:
            segments.append(f"{RESET}{HOT_PINK}│")
            i += 1
        elif i < filled:
            # Run of filled chars until target marker or end of filled section
            end = target_pos if i < target_pos < filled else filled
            segments.append(f"{bar_color}" + "█" * (end - i))
            i = end
        else:
            # Run of empty chars until target marker or end
            end = target_pos if target_pos > i else width
            segments.append(f"{DIM}{GRAY}" + "░" * (end - i))
            i = end

    return "".join(segments) + RESET


# ---------------------------------------------------------------------------
# Working directory
# ---------------------------------------------------------------------------

def shorten_dir(path: str) -> str:
    home = str(Path.home())
    if path.startswith(home):
        path = "~" + path[len(home):]
    if path.startswith("~/"):
        parts = path[2:].split("/")
        if len(parts) > 2:
            path = "~/…/" + "/".join(parts[-2:])
    return path


# ---------------------------------------------------------------------------
# Token velocity (EMA persisted per session)
# ---------------------------------------------------------------------------

EMA_ALPHA = 2 / 9  # N=8 turns


def update_velocity(session_id: str, total_tokens: int) -> tuple[int, float]:
    """Update EMA state and return (delta, ema)."""
    state_dir = Path.home() / ".claude"
    state_file = state_dir / f"statusline-state-{session_id}.json"

    prev_total = 0
    prev_ema = 0.0
    turn = 0

    try:
        state = json.loads(state_file.read_text())
        prev_total = state.get("total_tokens", 0)
        prev_ema = state.get("ema", 0.0)
        turn = state.get("turn", 0)
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        pass

    turn += 1
    delta = total_tokens - prev_total
    ema = float(delta) if turn <= 1 else EMA_ALPHA * delta + (1 - EMA_ALPHA) * prev_ema

    try:
        state_file.write_text(
            json.dumps({"turn": turn, "total_tokens": total_tokens, "ema": round(ema, 1)})
            + "\n"
        )
    except OSError:
        pass

    # Prune stale state files (>24h) — only on first turn to avoid repeated scans
    if turn == 1:
        try:
            cutoff = time.time() - 86400
            for f in state_dir.glob("statusline-state-*.json"):
                if f != state_file and f.stat().st_mtime < cutoff:
                    f.unlink(missing_ok=True)
        except OSError:
            pass

    return delta, ema


# ---------------------------------------------------------------------------
# Usage quota (5-hour and 7-day rolling windows)
# ---------------------------------------------------------------------------

USAGE_CACHE = Path("/tmp/claude-statusline-usage.json")
USAGE_CACHE_AGE = 60  # seconds


def get_oauth_token() -> str | None:
    """Read OAuth access token from keychain (macOS) or credentials file."""
    # macOS keychain
    try:
        result = subprocess.run(
            ["security", "find-generic-password", "-s", "Claude Code-credentials", "-w"],
            capture_output=True, text=True, timeout=3,
        )
        if result.returncode == 0:
            creds = json.loads(result.stdout.strip())
            token = creds.get("claudeAiOauth", {}).get("accessToken")
            if token:
                return token
    except (subprocess.TimeoutExpired, json.JSONDecodeError, FileNotFoundError, OSError):
        pass

    # Linux/WSL credentials file
    creds_file = Path.home() / ".claude" / ".credentials.json"
    try:
        creds = json.loads(creds_file.read_text())
        return creds.get("claudeAiOauth", {}).get("accessToken")
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        pass

    return None


def fetch_usage() -> dict | None:
    """Fetch usage from Anthropic API and cache it."""
    token = get_oauth_token()
    if not token:
        return None

    try:
        req = urllib.request.Request(
            "https://api.anthropic.com/api/oauth/usage",
            headers={
                "Authorization": f"Bearer {token}",
                "anthropic-beta": "oauth-2025-04-20",
            },
        )
        with urllib.request.urlopen(req, timeout=3) as resp:
            data = json.loads(resp.read())
    except Exception:
        return None

    if "five_hour" not in data or "seven_day" not in data:
        return None

    try:
        USAGE_CACHE.write_text(json.dumps(data))
    except OSError:
        pass

    return data


def get_usage(now: float) -> dict | None:
    """Return cached usage data, refreshing if stale.

    Falls back to stale cache if the API call fails.
    """
    cached = None
    try:
        if USAGE_CACHE.exists():
            cached = json.loads(USAGE_CACHE.read_text())
            age = now - USAGE_CACHE.stat().st_mtime
            if age <= USAGE_CACHE_AGE:
                return cached
    except (json.JSONDecodeError, OSError):
        pass

    return fetch_usage() or cached


def _reset_epoch(resets_at: str) -> float | None:
    """Parse an ISO timestamp to epoch seconds."""
    try:
        return datetime.fromisoformat(resets_at).timestamp()
    except ValueError:
        return None


def time_until_reset(resets_at: str, now: float) -> str | None:
    """Return a human-readable string like '2h13m' or '3d5h' until reset."""
    epoch = _reset_epoch(resets_at)
    if epoch is None:
        return None

    remaining = int(epoch - now)
    if remaining <= 0:
        return None

    days, remaining = divmod(remaining, 86400)
    hours, remaining = divmod(remaining, 3600)
    minutes = remaining // 60

    if days > 0:
        return f"{days}d{hours}h"
    if hours > 0:
        return f"{hours}h{minutes}m"
    return f"{minutes}m"


def pacing_target(resets_at: str, window_secs: int, now: float) -> float | None:
    """Compute what percentage of the window has elapsed (0-100)."""
    epoch = _reset_epoch(resets_at)
    if epoch is None:
        return None

    start_epoch = epoch - window_secs
    elapsed = max(0, min(now - start_epoch, window_secs))
    return elapsed / window_secs * 100


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def main() -> None:
    raw = sys.stdin.read().strip()
    if not raw:
        return

    try:
        data = json.loads(raw)
    except json.JSONDecodeError:
        return

    now = time.time()

    # --- Extract fields ---
    model = (data.get("model") or {}).get("display_name", "Unknown")
    session_id = data.get("session_id", "unknown")

    cw = data.get("context_window") or {}
    used_pct = cw.get("used_percentage")
    ctx_window_size = cw.get("context_window_size", 200000)

    cost_data = data.get("cost") or {}
    cost_usd = cost_data.get("total_cost_usd")
    duration_ms = cost_data.get("total_duration_ms")
    work_dir = (data.get("workspace") or {}).get("current_dir", "")

    # --- Context percentage ---
    # used_percentage is input-only and undercounts by ~10-15% vs Claude Code's
    # internal compaction threshold (which includes system prompts, tool
    # definitions, and reserved output buffer not exposed in the JSON).
    # current_usage.output_tokens is only the latest response (not cumulative),
    # so adding it barely moves the needle. We use used_percentage as-is.
    if used_pct is not None:
        ctx_pct = used_pct
        total_ctx = int(used_pct * ctx_window_size / 100) if ctx_window_size else 0
    else:
        ctx_pct = None
        total_ctx = 0

    # --- Token velocity ---
    total_tokens = cw.get("total_input_tokens", 0) + cw.get("total_output_tokens", 0)
    delta, ema = update_velocity(session_id, total_tokens)

    # --- Usage quota ---
    usage = get_usage(now)

    # === Build 3-column grid ===
    # Column 1: model/dir/duration  |  ctx bar
    # Column 2: cost/burn           |  5h bar
    # Column 3: velocity            |  7d bar

    # --- Row 1 cells ---
    r1c1_parts: list[str] = [f"{WHITE}{model}{RESET}"]
    if work_dir:
        r1c1_parts.append(shorten_dir(work_dir))
    if duration_ms is not None:
        r1c1_parts.append(f"{CYAN}{format_duration(duration_ms)}{RESET}")
    r1c1 = " ".join(r1c1_parts)

    r1c2 = ""
    if cost_usd is not None:
        r1c2 = f"${cost_usd:.2f}"
        if duration_ms is not None and int(duration_ms) // 1000 >= 10:
            hrs = duration_ms / 3_600_000
            burn_num = f"${cost_usd / hrs:.2f}" if hrs > 0 else "--"
            r1c2 += f" {dim('(')}{burn_num}{dim('/hr)')}"
        elif duration_ms is not None:
            r1c2 += f" {dim('(')}{dim('--')}{dim('/hr)')}"

    r1c3 = f"{dim('last')} {format_tok(delta)} {dim('avg')} {format_ema(ema)}{dim('/turn')}"

    # --- Row 2 cells ---
    if ctx_pct is not None:
        ctx_int = int(round(ctx_pct))
        bar = build_bar(ctx_pct, 15, green_thresh=60, yellow_thresh=85)
        cc = pct_color(ctx_pct, 60, 85)
        r2c1 = f"{dim('ctx')} {bar} {cc}{ctx_int}%{RESET} {dim('(')}{format_k(total_ctx)}{dim('/')}{format_k(ctx_window_size)}{dim(')')}"
    else:
        r2c1 = f"{dim('ctx')} {DIM}{GRAY}no data{RESET}"

    r2c2 = ""
    if usage and "five_hour" in usage:
        fh = usage["five_hour"]
        pct_5h = fh.get("utilization", 0)
        resets_5h = fh.get("resets_at", "")
        target = pacing_target(resets_5h, 5 * 3600, now)
        bar_5h = build_bar(pct_5h, 15, target)
        cc5 = pct_color(pct_5h)
        r2c2 = f"{dim('5h')} {bar_5h} {cc5}{int(round(pct_5h))}%{RESET}"
        ttl_5h = time_until_reset(resets_5h, now)
        if ttl_5h:
            r2c2 += f" {dim('⟳')}{ttl_5h}"

    r2c3 = ""
    if usage and "seven_day" in usage:
        sd = usage["seven_day"]
        pct_7d = sd.get("utilization", 0)
        resets_7d = sd.get("resets_at", "")
        target = pacing_target(resets_7d, 7 * 24 * 3600, now)
        bar_7d = build_bar(pct_7d, 15, target)
        cc7 = pct_color(pct_7d)
        r2c3 = f"{dim('7d')} {bar_7d} {cc7}{int(round(pct_7d))}%{RESET}"
        ttl_7d = time_until_reset(resets_7d, now)
        if ttl_7d:
            r2c3 += f" {dim('⟳')}{ttl_7d}"

    sys.stdout.write(render_grid([
        [r1c1, r1c2, r1c3],
        [r2c1, r2c2, r2c3],
    ]))


if __name__ == "__main__":
    main()