From 32f0cf527d2fae3ba69d8b9344c83aead49a6fdc Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 12:55:53 +0300
Subject: [PATCH 01/13] feat: add 5h/weekly usage percentages to statusline,
 remove sparkline

---
 scripts/statusline.py | 227 ++++++++++++++++++++++++------------------
 1 file changed, 130 insertions(+), 97 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index fcc9fdc..87559ba 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -15,7 +15,7 @@
 import sys
 import tempfile
 import time
-from datetime import datetime
+from datetime import datetime, timedelta
 from pathlib import Path
 
 # Cache configuration
@@ -275,17 +275,27 @@ def load_cost_cache() -> dict:
     return {}
 
 
-def save_cost_cache(daily_cost: str, session_cost: str, cwd: str) -> None:
+def save_cost_cache(
+    daily_cost: str,
+    session_cost: str,
+    cwd: str,
+    daily_cost_val: float = 0.0,
+    weekly_cost_val: float = 0.0,
+) -> None:
     """Save cost values to cache file.
 
     Args:
         daily_cost: Formatted daily cost string.
         session_cost: Formatted session cost string.
         cwd: Current working directory (for cache invalidation).
+        daily_cost_val: Raw daily cost value for percentage calculations.
+        weekly_cost_val: Raw weekly cost value for percentage calculations.
     """
     cache = {
         "daily_cost": daily_cost,
         "session_cost": session_cost,
+        "daily_cost_val": daily_cost_val,
+        "weekly_cost_val": weekly_cost_val,
         "timestamp": time.time(),
         "cwd": cwd,
     }
@@ -313,26 +323,33 @@ def is_cache_valid(cache: dict, cwd: str) -> bool:
     return age < COST_CACHE_TTL_SECONDS and cache_cwd == cwd
 
 
-def fetch_costs_raw(cwd: str) -> tuple[str, str]:
-    """Fetch daily and session costs from a single ccusage call.
+def fetch_costs_raw(cwd: str) -> tuple[str, str, float, float]:
+    """Fetch daily, session, and weekly costs from ccusage calls.
 
     Uses the `-i` flag which returns per-project breakdown that also includes
     daily totals, allowing both values to be extracted from one response.
+    Also fetches weekly data for usage percentage calculation.
 
     Args:
         cwd: Current working directory, used to identify the project.
 
     Returns:
-        Tuple of (daily_cost, session_cost) formatted strings like "$X.XX".
+        Tuple of (daily_cost, session_cost, daily_cost_val, weekly_cost_val)
+        where daily_cost/session_cost are formatted strings like "$X.XX"
+        and daily_cost_val/weekly_cost_val are raw float values.
     """
     today = datetime.now().strftime("%Y%m%d")
 
     # Convert cwd to project name format (replace / with -)
     project_name = cwd.replace("/", "-").replace("\\", "-") if cwd else ""
 
+    daily_cost_val = 0.0
+    weekly_cost_val = 0.0
+
     # Try bunx (bun) first, then npx - single call with -i flag gets both values
     for cmd in [["bunx", "ccusage@latest"], ["npx", "ccusage@latest"]]:
         try:
+            # Fetch today's costs
             result = subprocess.run(  # noqa: S603, S607
                 [*cmd, "daily", "--json", "--since", today, "-i"],
                 capture_output=True,
@@ -358,7 +375,19 @@ def fetch_costs_raw(cwd: str) -> tuple[str, str]:
                             )
                             session_cost = f"${total_cost:.2f}"
 
-                return daily_cost, session_cost
+                # Fetch weekly costs (last 7 days)
+                week_ago = (datetime.now() - timedelta(days=6)).strftime("%Y%m%d")
+                weekly_result = subprocess.run(  # noqa: S603, S607
+                    [*cmd, "daily", "--json", "--since", week_ago, "-i"],
+                    capture_output=True,
+                    text=True,
+                    timeout=30,
+                )
+                if weekly_result.returncode == 0 and weekly_result.stdout.strip():
+                    weekly_data = json.loads(weekly_result.stdout)
+                    weekly_cost_val = weekly_data.get("totals", {}).get("totalCost", 0)
+
+                return daily_cost, session_cost, daily_cost_val, weekly_cost_val
         except (
             subprocess.TimeoutExpired,
             FileNotFoundError,
@@ -367,7 +396,7 @@ def fetch_costs_raw(cwd: str) -> tuple[str, str]:
         ):
             continue
 
-    return "$0.00", "$0.00"
+    return "$0.00", "$0.00", 0.0, 0.0
 
 
 def _is_refresh_locked() -> bool:
@@ -414,7 +443,7 @@ def spawn_background_refresh(cwd: str) -> None:
     refresh_script = f"""
 import json, subprocess, time, sys
 from pathlib import Path
-from datetime import datetime
+from datetime import datetime, timedelta
 
 lock_file = Path({str(COST_REFRESH_LOCK)!r})
 cache_file = Path({str(COST_CACHE_FILE)!r})
@@ -425,10 +454,13 @@ def spawn_background_refresh(cwd: str) -> None:
     lock_file.write_text(str(time.time()))
 
     today = datetime.now().strftime("%Y%m%d")
+    week_ago = (datetime.now() - timedelta(days=6)).strftime("%Y%m%d")
     project_name = cwd.replace("/", "-").replace("\\\\", "-") if cwd else ""
 
     daily_cost = "$0.00"
     session_cost = "$0.00"
+    daily_cost_val = 0.0
+    weekly_cost_val = 0.0
 
     for cmd in [["bunx", "ccusage@latest"], ["npx", "ccusage@latest"]]:
         try:
@@ -447,6 +479,16 @@ def spawn_background_refresh(cwd: str) -> None:
                         if project_data and len(project_data) > 0:
                             total_cost = sum(e.get("totalCost", 0) for e in project_data)
                             session_cost = f"${{total_cost:.2f}}"
+
+                # Fetch weekly costs
+                weekly_result = subprocess.run(
+                    [*cmd, "daily", "--json", "--since", week_ago, "-i"],
+                    capture_output=True, text=True, timeout=30,
+                )
+                if weekly_result.returncode == 0 and weekly_result.stdout.strip():
+                    weekly_data = json.loads(weekly_result.stdout)
+                    weekly_cost_val = weekly_data.get("totals", {{}}).get("totalCost", 0)
+
                 break
         except Exception:
             continue
@@ -454,6 +496,8 @@ def spawn_background_refresh(cwd: str) -> None:
     cache = {{
         "daily_cost": daily_cost,
         "session_cost": session_cost,
+        "daily_cost_val": daily_cost_val,
+        "weekly_cost_val": weekly_cost_val,
         "timestamp": time.time(),
         "cwd": cwd,
     }}
@@ -473,7 +517,7 @@ def spawn_background_refresh(cwd: str) -> None:
         debug_log(f"Failed to spawn background refresh: {e}")
 
 
-def get_costs_cached(cwd: str) -> tuple[str, str]:
+def get_costs_cached(cwd: str) -> tuple[str, str, float, float]:
     """Get daily and session costs with non-blocking cache refresh.
 
     Returns cached values immediately. If the cache is expired, returns stale
@@ -486,7 +530,7 @@ def get_costs_cached(cwd: str) -> tuple[str, str]:
         cwd: Current working directory for session cost lookup.
 
     Returns:
-        Tuple of (daily_cost, session_cost) formatted strings like "$X.XX".
+        Tuple of (daily_cost, session_cost, daily_cost_val, weekly_cost_val).
     """
     cache = load_cost_cache()
 
@@ -494,24 +538,33 @@ def get_costs_cached(cwd: str) -> tuple[str, str]:
         debug_log(
             f"Using cached costs (age: {time.time() - cache.get('timestamp', 0):.1f}s)"
         )
-        return cache.get("daily_cost", "$0.00"), cache.get("session_cost", "$0.00")
+        return (
+            cache.get("daily_cost", "$0.00"),
+            cache.get("session_cost", "$0.00"),
+            cache.get("daily_cost_val", 0.0),
+            cache.get("weekly_cost_val", 0.0),
+        )
 
     # Cache is stale or missing - return immediately with stale/placeholder values
     if cache:
         stale_daily = cache.get("daily_cost", "$...")
         stale_session = cache.get("session_cost", "$...")
+        stale_daily_val = cache.get("daily_cost_val", 0.0)
+        stale_weekly_val = cache.get("weekly_cost_val", 0.0)
         debug_log(
             f"Cache expired, returning stale values: daily={stale_daily}, session={stale_session}"
         )
     else:
         stale_daily = "$..."
         stale_session = "$..."
+        stale_daily_val = 0.0
+        stale_weekly_val = 0.0
         debug_log("No cache exists, returning placeholders")
 
     # Spawn background refresh (fire and forget)
     spawn_background_refresh(cwd)
 
-    return stale_daily, stale_session
+    return stale_daily, stale_session, stale_daily_val, stale_weekly_val
 
 
 def get_claude_version() -> str:
@@ -534,79 +587,14 @@ def get_claude_version() -> str:
     return "v?"
 
 
-def generate_sparkline(durations: list[float]) -> str:
-    """Generate a sparkline visualization from duration values.
-
-    Maps each duration to a block character based on its relative position
-    between the min and max values in the array.
-
-    Args:
-        durations: List of duration values in seconds.
-
-    Returns:
-        Sparkline string using block characters (e.g., "▁▃▅▂█").
-    """
-    if not durations:
-        return ""
-
-    # Sparkline characters from lowest to highest
-    spark_chars = "▁▂▃▄▅▆▇█"
-    max_index = len(spark_chars) - 1
-
-    min_val = min(durations)
-    max_val = max(durations)
-    value_range = max_val - min_val
-
-    sparkline = ""
-    for duration in durations:
-        if value_range == 0:
-            # All values are equal - use middle height
-            index = max_index // 2
-        else:
-            # Map value to 0-7 index based on position in range
-            normalized = (duration - min_val) / value_range
-            index = int(normalized * max_index)
-            # Clamp to valid range
-            index = max(0, min(max_index, index))
-        sparkline += spark_chars[index]
-
-    return sparkline
-
-
-def get_duration_history() -> list[float]:
-    """Get the history of turn durations for sparkline visualization.
-
-    Reads from the JSON file maintained by the stop hook.
-
-    Returns:
-        List of duration values in seconds, or empty list if not available.
-    """
-    state_dir = (
-        Path(os.environ.get("CLAUDE_PROJECT_DIR", Path.cwd())) / ".claude" / "state"
-    )
-    history_file = state_dir / "turn_durations.json"
-
-    if not history_file.exists():
-        return []
-
-    try:
-        data = json.loads(history_file.read_text(encoding="utf-8"))
-        durations = data.get("durations", [])
-        # Validate and convert to floats
-        return [float(d) for d in durations if isinstance(d, int | float)]
-    except (json.JSONDecodeError, ValueError, OSError):
-        return []
-
-
 def get_turn_duration() -> str | None:
-    """Get the duration of the last completed turn with sparkline visualization.
+    """Get the duration of the last completed turn.
 
     Reads from the state file written by the stop hook which calculates
-    duration from UserPromptSubmit to Stop events. Also includes a sparkline
-    showing the trend of the last 10 turn durations.
+    duration from UserPromptSubmit to Stop events.
 
     Returns:
-        Formatted duration string like "▁▃▅▂█ 45s", or None if not available.
+        Formatted duration string like "45s" or "1m 23s", or None if not available.
     """
     state_dir = (
         Path(os.environ.get("CLAUDE_PROJECT_DIR", Path.cwd())) / ".claude" / "state"
@@ -619,16 +607,6 @@ def get_turn_duration() -> str | None:
     try:
         duration_str = duration_file.read_text(encoding="utf-8").strip()
         if duration_str:
-            # Get sparkline from duration history
-            durations = get_duration_history()
-            sparkline = generate_sparkline(durations)
-
-            if sparkline:
-                # Color the sparkline with coral/salmon (RGB true color)
-                # Fallback-safe: Windows Terminal supports RGB, older terminals ignore
-                CORAL = "\033[38;2;255;127;80m"
-                RESET = "\033[0m"
-                return f"{CORAL}{sparkline}{RESET} {duration_str}"
             return duration_str
     except OSError:
         pass
@@ -636,6 +614,54 @@ def get_turn_duration() -> str | None:
     return None
 
 
+def format_usage_percentages(
+    daily_cost_val: float,
+    weekly_cost_val: float,
+    daily_limit: float,
+    weekly_limit: float,
+) -> str:
+    """Format 5h and weekly usage as colored percentage strings.
+
+    Uses daily spend as approximation for 5h usage (ccusage gives daily granularity).
+
+    Args:
+        daily_cost_val: Raw daily cost value.
+        weekly_cost_val: Raw weekly cost value (sum of last 7 days).
+        daily_limit: Daily spend limit in dollars.
+        weekly_limit: Weekly spend limit in dollars (daily_limit * 7).
+
+    Returns:
+        Formatted string like "5h 89% · weekly 91%" with ANSI color codes.
+    """
+    reset = "\033[0m"
+
+    def color_for_pct(pct: float) -> str:
+        if pct > 75:
+            return "\033[31m"  # Red
+        elif pct > 50:
+            return "\033[33m"  # Yellow
+        return "\033[32m"  # Green
+
+    if daily_limit > 0:
+        five_h_pct = (daily_cost_val / daily_limit) * 100
+    else:
+        five_h_pct = 0.0
+
+    if weekly_limit > 0:
+        weekly_pct = (weekly_cost_val / weekly_limit) * 100
+    else:
+        weekly_pct = 0.0
+
+    five_h_color = color_for_pct(five_h_pct)
+    weekly_color = color_for_pct(weekly_pct)
+
+    return (
+        f"{five_h_color}5h {five_h_pct:.0f}%{reset}"
+        f" · "
+        f"{weekly_color}weekly {weekly_pct:.0f}%{reset}"
+    )
+
+
 def main() -> None:
     """Main entry point."""
     # Clear debug log at start
@@ -678,7 +704,9 @@ def main() -> None:
     full_cwd = os.getcwd()
 
     # Get daily and session costs (with caching for fast statusline refresh)
-    daily_cost, session_cost = get_costs_cached(full_cwd)
+    daily_cost, session_cost, daily_cost_val, weekly_cost_val = get_costs_cached(
+        full_cwd
+    )
 
     # Get Claude version
     claude_version = get_claude_version()
@@ -702,22 +730,27 @@ def main() -> None:
     # Get turn duration if available
     turn_duration = get_turn_duration()
 
-    # Format costs: extract numbers and combine as "💰 $Y.YY (🎯 $X.XX)"
-    # where daily_cost is first (larger), session_cost in parentheses
-    # session_cost format: "$X.XX"
-    # daily_cost format: "$Y.YY"
+    # Format costs: extract numbers and combine as "costs (session)"
     daily_amount = daily_cost
     session_amount = session_cost
     cost_display = f"{GREEN}💰 {daily_amount} (🎯 {session_amount}){RESET}"
 
+    # Calculate usage percentages
+    daily_limit = float(os.environ.get("CLAUDE_DAILY_LIMIT", "200"))
+    weekly_limit = daily_limit * 7
+
+    usage_display = format_usage_percentages(
+        daily_cost_val, weekly_cost_val, daily_limit, weekly_limit
+    )
+
     # Output statusline (print is required for statusline output)
     # Row 1 (static per session): version, model, project dir, git branch
     cwd_display = f"{SHINY_AQUA}📁 {cwd}{RESET}"
     sys.stdout.write(
         f"{BLUE}{claude_version}{RESET} | {SHINY_AQUA}🤖 {raw_model}{RESET} | {cwd_display} | {YELLOW}{git_status}{RESET}\n"
     )
-    # Row 2 (dynamic metrics): costs, context bar, turn duration
-    row2_parts = [cost_display, context_info]
+    # Row 2 (dynamic metrics): costs, context bar, usage percentages, turn duration
+    row2_parts = [cost_display, context_info, usage_display]
     if turn_duration:
         row2_parts.append(f"{YELLOW}⏱️ {turn_duration}{RESET}")
     sys.stdout.write(" | ".join(row2_parts) + "\n")

From 9e98be08b5b9cb7b70a6c396162488bc674b810b Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 13:03:07 +0300
Subject: [PATCH 02/13] fix: shorten context bar width from 20 to 10 chars

---
 scripts/statusline.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 87559ba..3ace5b9 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -45,9 +45,9 @@ def create_progress_bar(usage_rate: float, used_tokens: int, limit_tokens: int)
     """Create a colored progress bar for context usage."""
     percentage = max(0, min(100, int(usage_rate)))
 
-    # Calculate filled blocks (20 total)
-    filled_blocks = percentage * 20 // 100
-    empty_blocks = 20 - filled_blocks
+    # Calculate filled blocks (10 total)
+    filled_blocks = percentage * 10 // 100
+    empty_blocks = 10 - filled_blocks
 
     # Build bar using Unicode block characters
     filled_part = "█" * filled_blocks

From 5891120620c2a17bed6d9a33fd8e3979735d7343 Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 13:04:02 +0300
Subject: [PATCH 03/13] fix: remove session cost from statusline

---
 scripts/statusline.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 3ace5b9..15dce2f 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -730,10 +730,9 @@ def main() -> None:
     # Get turn duration if available
     turn_duration = get_turn_duration()
 
-    # Format costs: extract numbers and combine as "costs (session)"
+    # Format costs: daily cost only
     daily_amount = daily_cost
-    session_amount = session_cost
-    cost_display = f"{GREEN}💰 {daily_amount} (🎯 {session_amount}){RESET}"
+    cost_display = f"{GREEN}💰 {daily_amount}{RESET}"
 
     # Calculate usage percentages
     daily_limit = float(os.environ.get("CLAUDE_DAILY_LIMIT", "200"))

From 0a69c9f1dad566131efffcc1feb5206aa1b3f2fb Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 13:07:25 +0300
Subject: [PATCH 04/13] feat: responsive statusline with truncation for narrow
 terminals

---
 scripts/statusline.py | 99 +++++++++++++++++++++++++++++++++++--------
 1 file changed, 82 insertions(+), 17 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 15dce2f..146e0cd 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -241,8 +241,30 @@ def shorten_cwd(full_path: str) -> str:
     return f".../{Path(full_path).name}"
 
 
+def truncate_str(s: str, max_len: int) -> str:
+    """Truncate a string to max_len, showing start + ellipsis + end if too long."""
+    if len(s) <= max_len:
+        return s
+    if max_len <= 3:
+        return s[:max_len]
+    # Show roughly first 40% and last 50% with ellipsis
+    head = max_len * 2 // 5
+    tail = max_len - head - 1
+    return s[:head] + "\u2026" + s[len(s) - tail :]
+
+
+def get_terminal_width() -> int:
+    """Get terminal width, defaulting to 120 if detection fails."""
+    try:
+        import shutil
+
+        return shutil.get_terminal_size(fallback=(120, 24)).columns
+    except Exception:
+        return 120
+
+
 def get_git_branch() -> str:
-    """Get the current git branch."""
+    """Get the current git branch (raw name, no emoji)."""
     try:
         result = subprocess.run(
             ["git", "branch", "--show-current"],  # noqa: S607, S603
@@ -251,13 +273,10 @@ def get_git_branch() -> str:
             timeout=5,
         )
         if result.returncode == 0 and result.stdout.strip():
-            branch = result.stdout.strip()
-            if len(branch) > 60:
-                branch = branch[:60] + "..."
-            return f"🌿 {branch} ⚡"
+            return result.stdout.strip()
     except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
         pass
-    return "🌿 no-git"
+    return "no-git"
 
 
 def load_cost_cache() -> dict:
@@ -721,8 +740,8 @@ def main() -> None:
         progress_bar = create_progress_bar(0, 0, max_context)
         context_info = f"🧠 {progress_bar}"
 
-    # Get git status
-    git_status = get_git_branch()
+    # Get git branch (raw name)
+    branch_raw = get_git_branch()
 
     # Get shortened CWD
     cwd = shorten_cwd(os.getcwd())
@@ -732,7 +751,7 @@ def main() -> None:
 
     # Format costs: daily cost only
     daily_amount = daily_cost
-    cost_display = f"{GREEN}💰 {daily_amount}{RESET}"
+    cost_display = f"{GREEN}\U0001f4b0 {daily_amount}{RESET}"
 
     # Calculate usage percentages
     daily_limit = float(os.environ.get("CLAUDE_DAILY_LIMIT", "200"))
@@ -742,16 +761,62 @@ def main() -> None:
         daily_cost_val, weekly_cost_val, daily_limit, weekly_limit
     )
 
-    # Output statusline (print is required for statusline output)
-    # Row 1 (static per session): version, model, project dir, git branch
-    cwd_display = f"{SHINY_AQUA}📁 {cwd}{RESET}"
+    # Detect terminal width for responsive layout
+    term_width = get_terminal_width()
+
+    # --- Row 1: version | model | dir | branch ---
+    # Apply initial truncation limits
+    dir_max = 25
+    branch_max = 20
+    cwd_trunc = truncate_str(cwd, dir_max)
+    branch_trunc = truncate_str(branch_raw, branch_max)
+
+    # Estimate visible length of row 1 (exclude ANSI codes, emoji ~2 chars each)
+    # Format: "vX.X.X | 🤖 model | 📁 dir | 🌿 branch ⚡"
+    def _row1_visible_len(d: str, b: str) -> int:
+        return (
+            len(claude_version)
+            + 3  # " | "
+            + 2
+            + len(raw_model)  # "🤖 " + model
+            + 3  # " | "
+            + 2
+            + len(d)  # "📁 " + dir
+            + 3  # " | "
+            + 2
+            + len(b)
+            + 2  # "🌿 " + branch + " ⚡"
+        )
+
+    # Progressive shortening if row 1 exceeds terminal width
+    est_len = _row1_visible_len(cwd_trunc, branch_trunc)
+    if est_len > term_width:
+        # First: shorten branch more aggressively
+        branch_trunc = truncate_str(
+            branch_raw, max(10, branch_max - (est_len - term_width))
+        )
+        est_len = _row1_visible_len(cwd_trunc, branch_trunc)
+    if est_len > term_width:
+        # Second: shorten directory more aggressively
+        cwd_trunc = truncate_str(cwd, max(10, dir_max - (est_len - term_width)))
+
+    git_status = f"\U0001f33f {branch_trunc} \u26a1"
+    cwd_display = f"{SHINY_AQUA}\U0001f4c1 {cwd_trunc}{RESET}"
+
+    # Output Row 1
     sys.stdout.write(
-        f"{BLUE}{claude_version}{RESET} | {SHINY_AQUA}🤖 {raw_model}{RESET} | {cwd_display} | {YELLOW}{git_status}{RESET}\n"
+        f"{BLUE}{claude_version}{RESET} | {SHINY_AQUA}\U0001f916 {raw_model}{RESET} | {cwd_display} | {YELLOW}{git_status}{RESET}\n"
     )
-    # Row 2 (dynamic metrics): costs, context bar, usage percentages, turn duration
-    row2_parts = [cost_display, context_info, usage_display]
-    if turn_duration:
-        row2_parts.append(f"{YELLOW}⏱️ {turn_duration}{RESET}")
+
+    # --- Row 2: context bar | usage | cost | duration ---
+    # Priority order: context_info, usage_display, cost_display, duration
+    # Build parts with priority-based dropping for narrow terminals
+    row2_parts = [context_info, usage_display]
+    if term_width >= 80:
+        row2_parts.append(cost_display)
+    if term_width >= 60 and turn_duration:
+        row2_parts.append(f"{YELLOW}\u23f1\ufe0f {turn_duration}{RESET}")
+
     sys.stdout.write(" | ".join(row2_parts) + "\n")
 
 

From 38ccccc96db7d4c8c6a452d7f0623aef47204587 Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 13:17:29 +0300
Subject: [PATCH 05/13] fix: ensure row 2 always displays in narrow terminals

---
 scripts/statusline.py | 99 +++++++++++++++++++++++++++++++++++++++----
 1 file changed, 90 insertions(+), 9 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 146e0cd..3987875 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -254,11 +254,21 @@ def truncate_str(s: str, max_len: int) -> str:
 
 
 def get_terminal_width() -> int:
-    """Get terminal width, defaulting to 120 if detection fails."""
+    """Get terminal width, defaulting to 120 if detection fails.
+
+    In Claude Code's statusline context, shutil.get_terminal_size() often
+    returns unreasonably small values because it's not a real TTY.
+    We treat anything below 40 columns as unreliable and default to 120.
+    """
     try:
         import shutil
 
-        return shutil.get_terminal_size(fallback=(120, 24)).columns
+        width = shutil.get_terminal_size(fallback=(120, 24)).columns
+        # In non-TTY contexts (like Claude Code statusline), the detected
+        # width can be 0 or very small. Use a conservative default instead.
+        if width < 40:
+            return 120
+        return width
     except Exception:
         return 120
 
@@ -638,6 +648,7 @@ def format_usage_percentages(
     weekly_cost_val: float,
     daily_limit: float,
     weekly_limit: float,
+    compact: bool = False,
 ) -> str:
     """Format 5h and weekly usage as colored percentage strings.
 
@@ -648,6 +659,7 @@ def format_usage_percentages(
         weekly_cost_val: Raw weekly cost value (sum of last 7 days).
         daily_limit: Daily spend limit in dollars.
         weekly_limit: Weekly spend limit in dollars (daily_limit * 7).
+        compact: If True, use shorter labels (e.g., "5h 2%·w 60%").
 
     Returns:
         Formatted string like "5h 89% · weekly 91%" with ANSI color codes.
@@ -674,6 +686,13 @@ def color_for_pct(pct: float) -> str:
     five_h_color = color_for_pct(five_h_pct)
     weekly_color = color_for_pct(weekly_pct)
 
+    if compact:
+        return (
+            f"{five_h_color}5h {five_h_pct:.0f}%{reset}"
+            f"\u00b7"
+            f"{weekly_color}w {weekly_pct:.0f}%{reset}"
+        )
+
     return (
         f"{five_h_color}5h {five_h_pct:.0f}%{reset}"
         f" · "
@@ -809,13 +828,75 @@ def _row1_visible_len(d: str, b: str) -> int:
     )
 
     # --- Row 2: context bar | usage | cost | duration ---
-    # Priority order: context_info, usage_display, cost_display, duration
-    # Build parts with priority-based dropping for narrow terminals
-    row2_parts = [context_info, usage_display]
-    if term_width >= 80:
-        row2_parts.append(cost_display)
-    if term_width >= 60 and turn_duration:
-        row2_parts.append(f"{YELLOW}\u23f1\ufe0f {turn_duration}{RESET}")
+    # Row 2 must ALWAYS show something. At minimum: context percentage and usage.
+    # Progressive compaction for narrow terminals instead of hiding elements.
+    if term_width >= 100:
+        # Full layout: context bar | usage | cost | duration
+        row2_parts = [context_info, usage_display, cost_display]
+        if turn_duration:
+            row2_parts.append(f"{YELLOW}\u23f1\ufe0f {turn_duration}{RESET}")
+    elif term_width >= 80:
+        # Medium: context bar | usage | cost (no duration)
+        row2_parts = [context_info, usage_display, cost_display]
+    elif term_width >= 60:
+        # Compact: context bar | compact usage (drop cost, duration)
+        compact_usage = format_usage_percentages(
+            daily_cost_val,
+            weekly_cost_val,
+            daily_limit,
+            weekly_limit,
+            compact=True,
+        )
+        row2_parts = [context_info, compact_usage]
+    else:
+        # Minimal: just percentage + compact usage (drop bar, emoji, cost, duration)
+        # Extract percentage from context_info by recalculating
+        compact_usage = format_usage_percentages(
+            daily_cost_val,
+            weekly_cost_val,
+            daily_limit,
+            weekly_limit,
+            compact=True,
+        )
+        # Build a minimal context display (just percentage, no bar)
+        session_id = input_data.get("session_id", "")
+        max_ctx = 200000
+        if "1M" in raw_model:
+            max_ctx = 1_000_000
+        # Try to get actual percentage from session data
+        ctx_pct = "0%"
+        if session_id:
+            session_file = find_session_file(
+                session_id,
+                input_data.get("cwd")
+                or input_data.get("workspace", {}).get("current_dir", ""),
+            )
+            if session_file:
+                try:
+                    last_reset_line = 0
+                    lines_list: list[str] = []
+                    with open(session_file, "r", encoding="utf-8") as f:
+                        for i, line in enumerate(f, 1):
+                            lines_list.append(line)
+                            if '"/clear"' in line or '"/compact"' in line:
+                                last_reset_line = i
+                    for line in lines_list[last_reset_line:]:
+                        try:
+                            entry = json.loads(line)
+                            usage_data = entry.get("message", {}).get("usage", {})
+                            if usage_data:
+                                total_in = (
+                                    usage_data.get("input_tokens", 0)
+                                    + usage_data.get("cache_read_input_tokens", 0)
+                                    + usage_data.get("cache_creation_input_tokens", 0)
+                                )
+                                if total_in > 0:
+                                    ctx_pct = f"{total_in * 100 / max_ctx:.0f}%"
+                        except json.JSONDecodeError:
+                            continue
+                except OSError:
+                    pass
+        row2_parts = [f"{ctx_pct}", compact_usage]
 
     sys.stdout.write(" | ".join(row2_parts) + "\n")
 

From cc6aee4ee31cfe8f5fda307ad53ba2e48d8c6e4d Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 13:37:47 +0300
Subject: [PATCH 06/13] feat: use real Anthropic OAuth usage API for 5h/weekly
 percentages

---
 scripts/statusline.py | 293 ++++++++++++++++++++++++++++++++++++------
 1 file changed, 253 insertions(+), 40 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 3987875..2ff0ab0 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -23,6 +23,12 @@
 COST_CACHE_FILE = Path(tempfile.gettempdir()) / "statusline_cost_cache.json"
 COST_REFRESH_LOCK = Path(tempfile.gettempdir()) / "statusline_cost_refresh.lock"  # noqa: S108
 
+# Usage API cache (separate from cost cache — different TTL)
+USAGE_CACHE_TTL_SECONDS = 120  # 2 minutes — aggressive but avoids 429s
+USAGE_CACHE_FILE = Path(tempfile.gettempdir()) / "statusline_usage_cache.json"
+USAGE_REFRESH_LOCK = Path(tempfile.gettempdir()) / "statusline_usage_refresh.lock"  # noqa: S108
+USAGE_429_BACKOFF_SECONDS = 300  # 5 minutes on rate limit
+
 # Force UTF-8 output on Windows (fixes emoji encoding errors)
 if sys.platform == "win32":
     sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
@@ -546,6 +552,227 @@ def spawn_background_refresh(cwd: str) -> None:
         debug_log(f"Failed to spawn background refresh: {e}")
 
 
+def load_usage_cache() -> dict:
+    """Load usage API cache from file."""
+    try:
+        if USAGE_CACHE_FILE.exists():
+            return json.loads(USAGE_CACHE_FILE.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        pass
+    return {}
+
+
+def save_usage_cache(
+    five_hour_pct: float | None,
+    seven_day_pct: float | None,
+    error_429: bool = False,
+) -> None:
+    """Save usage API values to cache file."""
+    cache: dict = {
+        "five_hour_pct": five_hour_pct,
+        "seven_day_pct": seven_day_pct,
+        "timestamp": time.time(),
+    }
+    if error_429:
+        cache["backoff_until"] = time.time() + USAGE_429_BACKOFF_SECONDS
+    try:
+        USAGE_CACHE_FILE.write_text(json.dumps(cache), encoding="utf-8")
+    except OSError:
+        pass
+
+
+def is_usage_cache_valid(cache: dict) -> bool:
+    """Check if usage cache is still valid."""
+    if not cache:
+        return False
+    # If in 429 backoff, cache is valid until backoff expires
+    backoff_until = cache.get("backoff_until", 0)
+    if backoff_until and time.time() < backoff_until:
+        return True
+    cache_time = cache.get("timestamp", 0)
+    age = time.time() - cache_time
+    return age < USAGE_CACHE_TTL_SECONDS
+
+
+def _is_usage_refresh_locked() -> bool:
+    """Check if a usage background refresh is already running."""
+    try:
+        if USAGE_REFRESH_LOCK.exists():
+            lock_age = time.time() - USAGE_REFRESH_LOCK.stat().st_mtime
+            if lock_age < 60:
+                return True
+            USAGE_REFRESH_LOCK.unlink(missing_ok=True)
+    except OSError:
+        pass
+    return False
+
+
+def fetch_usage_data() -> tuple[float | None, float | None]:
+    """Fetch usage data from Anthropic OAuth usage API.
+
+    Reads OAuth token from ~/.claude/.credentials.json and calls the API.
+
+    Returns:
+        Tuple of (five_hour_pct, seven_day_pct) or (None, None) on failure.
+    """
+    import urllib.request
+    import urllib.error
+
+    # Read OAuth token
+    creds_path = Path.home() / ".claude" / ".credentials.json"
+    if not creds_path.exists():
+        # Windows fallback
+        appdata = os.environ.get("APPDATA", "")
+        if appdata:
+            creds_path = Path(appdata) / "claude" / ".credentials.json"
+    if not creds_path.exists():
+        debug_log("No credentials file found for usage API")
+        return None, None
+
+    try:
+        creds = json.loads(creds_path.read_text(encoding="utf-8"))
+        token = creds.get("claudeAiOauth", {}).get("accessToken")
+        if not token:
+            debug_log("No OAuth access token found in credentials")
+            return None, None
+    except (json.JSONDecodeError, OSError) as e:
+        debug_log(f"Failed to read credentials: {e}")
+        return None, None
+
+    # Call the API
+    url = "https://api.anthropic.com/api/oauth/usage"
+    req = urllib.request.Request(  # noqa: S310
+        url,
+        headers={
+            "Authorization": f"Bearer {token}",
+            "anthropic-beta": "oauth-2025-04-20",
+        },
+    )
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
+            data = json.loads(resp.read().decode("utf-8"))
+        five_hour = data.get("five_hour", {}).get("utilization")
+        seven_day = data.get("seven_day", {}).get("utilization")
+        debug_log(f"Usage API: 5h={five_hour}%, 7d={seven_day}%")
+        return five_hour, seven_day
+    except urllib.error.HTTPError as e:
+        if e.code == 429:
+            debug_log("Usage API rate limited (429), backing off 5 minutes")
+            save_usage_cache(None, None, error_429=True)
+        else:
+            debug_log(f"Usage API HTTP error: {e.code}")
+        return None, None
+    except (urllib.error.URLError, OSError, json.JSONDecodeError) as e:
+        debug_log(f"Usage API error: {e}")
+        return None, None
+
+
+def spawn_usage_background_refresh() -> None:
+    """Spawn a background process to refresh the usage API cache."""
+    if _is_usage_refresh_locked():
+        debug_log("Usage background refresh already running, skipping")
+        return
+
+    debug_log("Spawning usage background refresh process")
+
+    refresh_script = f"""
+import json, time, sys, os
+import urllib.request
+import urllib.error
+from pathlib import Path
+
+lock_file = Path({str(USAGE_REFRESH_LOCK)!r})
+cache_file = Path({str(USAGE_CACHE_FILE)!r})
+BACKOFF_SECONDS = {USAGE_429_BACKOFF_SECONDS}
+
+try:
+    lock_file.write_text(str(time.time()))
+
+    creds_path = Path.home() / ".claude" / ".credentials.json"
+    if not creds_path.exists():
+        appdata = os.environ.get("APPDATA", "")
+        if appdata:
+            creds_path = Path(appdata) / "claude" / ".credentials.json"
+
+    five_hour_pct = None
+    seven_day_pct = None
+    error_429 = False
+
+    if creds_path.exists():
+        creds = json.loads(creds_path.read_text(encoding="utf-8"))
+        token = creds.get("claudeAiOauth", {{}}).get("accessToken")
+        if token:
+            url = "https://api.anthropic.com/api/oauth/usage"
+            req = urllib.request.Request(
+                url,
+                headers={{
+                    "Authorization": f"Bearer {{token}}",
+                    "anthropic-beta": "oauth-2025-04-20",
+                }},
+            )
+            try:
+                with urllib.request.urlopen(req, timeout=10) as resp:
+                    data = json.loads(resp.read().decode("utf-8"))
+                five_hour_pct = data.get("five_hour", {{}}).get("utilization")
+                seven_day_pct = data.get("seven_day", {{}}).get("utilization")
+            except urllib.error.HTTPError as e:
+                if e.code == 429:
+                    error_429 = True
+            except Exception:
+                pass
+
+    cache = {{
+        "five_hour_pct": five_hour_pct,
+        "seven_day_pct": seven_day_pct,
+        "timestamp": time.time(),
+    }}
+    if error_429:
+        cache["backoff_until"] = time.time() + BACKOFF_SECONDS
+    cache_file.write_text(json.dumps(cache))
+finally:
+    lock_file.unlink(missing_ok=True)
+"""
+
+    try:
+        subprocess.Popen(  # noqa: S603
+            [sys.executable, "-c", refresh_script],  # noqa: S603
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+            start_new_session=True,
+        )
+    except OSError as e:
+        debug_log(f"Failed to spawn usage background refresh: {e}")
+
+
+def get_usage_cached() -> tuple[float | None, float | None]:
+    """Get usage percentages with non-blocking cache refresh.
+
+    Returns cached values immediately. If cache is expired, returns stale
+    values and spawns a background refresh.
+
+    Returns:
+        Tuple of (five_hour_pct, seven_day_pct) — None means data unavailable.
+    """
+    cache = load_usage_cache()
+
+    if is_usage_cache_valid(cache):
+        debug_log(
+            f"Using cached usage (age: {time.time() - cache.get('timestamp', 0):.1f}s)"
+        )
+        return cache.get("five_hour_pct"), cache.get("seven_day_pct")
+
+    # Cache stale or missing — return stale values and refresh in background
+    stale_5h = cache.get("five_hour_pct") if cache else None
+    stale_7d = cache.get("seven_day_pct") if cache else None
+    debug_log(
+        f"Usage cache expired/missing, stale values: 5h={stale_5h}, 7d={stale_7d}"
+    )
+
+    spawn_usage_background_refresh()
+
+    return stale_5h, stale_7d
+
+
 def get_costs_cached(cwd: str) -> tuple[str, str, float, float]:
     """Get daily and session costs with non-blocking cache refresh.
 
@@ -644,25 +871,21 @@ def get_turn_duration() -> str | None:
 
 
 def format_usage_percentages(
-    daily_cost_val: float,
-    weekly_cost_val: float,
-    daily_limit: float,
-    weekly_limit: float,
+    five_hour_pct: float | None,
+    seven_day_pct: float | None,
     compact: bool = False,
 ) -> str:
     """Format 5h and weekly usage as colored percentage strings.
 
-    Uses daily spend as approximation for 5h usage (ccusage gives daily granularity).
+    Uses real utilization values from the Anthropic OAuth usage API.
 
     Args:
-        daily_cost_val: Raw daily cost value.
-        weekly_cost_val: Raw weekly cost value (sum of last 7 days).
-        daily_limit: Daily spend limit in dollars.
-        weekly_limit: Weekly spend limit in dollars (daily_limit * 7).
+        five_hour_pct: 5-hour utilization percentage (0-100) from API, or None if unavailable.
+        seven_day_pct: 7-day utilization percentage (0-100) from API, or None if unavailable.
         compact: If True, use shorter labels (e.g., "5h 2%·w 60%").
 
     Returns:
-        Formatted string like "5h 89% · weekly 91%" with ANSI color codes.
+        Formatted string like "5h 6% · weekly 1%" with ANSI color codes.
     """
     reset = "\033[0m"
 
@@ -673,30 +896,31 @@ def color_for_pct(pct: float) -> str:
             return "\033[33m"  # Yellow
         return "\033[32m"  # Green
 
-    if daily_limit > 0:
-        five_h_pct = (daily_cost_val / daily_limit) * 100
+    if five_hour_pct is not None:
+        five_h_str = f"{five_hour_pct:.0f}%"
+        five_h_color = color_for_pct(five_hour_pct)
     else:
-        five_h_pct = 0.0
+        five_h_str = "\u2026%"
+        five_h_color = "\033[90m"  # Gray for placeholder
 
-    if weekly_limit > 0:
-        weekly_pct = (weekly_cost_val / weekly_limit) * 100
+    if seven_day_pct is not None:
+        weekly_str = f"{seven_day_pct:.0f}%"
+        weekly_color = color_for_pct(seven_day_pct)
     else:
-        weekly_pct = 0.0
-
-    five_h_color = color_for_pct(five_h_pct)
-    weekly_color = color_for_pct(weekly_pct)
+        weekly_str = "\u2026%"
+        weekly_color = "\033[90m"  # Gray for placeholder
 
     if compact:
         return (
-            f"{five_h_color}5h {five_h_pct:.0f}%{reset}"
+            f"{five_h_color}5h {five_h_str}{reset}"
             f"\u00b7"
-            f"{weekly_color}w {weekly_pct:.0f}%{reset}"
+            f"{weekly_color}w {weekly_str}{reset}"
         )
 
     return (
-        f"{five_h_color}5h {five_h_pct:.0f}%{reset}"
-        f" · "
-        f"{weekly_color}weekly {weekly_pct:.0f}%{reset}"
+        f"{five_h_color}5h {five_h_str}{reset}"
+        f" \u00b7 "
+        f"{weekly_color}weekly {weekly_str}{reset}"
     )
 
 
@@ -772,13 +996,10 @@ def main() -> None:
     daily_amount = daily_cost
     cost_display = f"{GREEN}\U0001f4b0 {daily_amount}{RESET}"
 
-    # Calculate usage percentages
-    daily_limit = float(os.environ.get("CLAUDE_DAILY_LIMIT", "200"))
-    weekly_limit = daily_limit * 7
+    # Get real usage percentages from Anthropic OAuth API
+    five_hour_pct, seven_day_pct = get_usage_cached()
 
-    usage_display = format_usage_percentages(
-        daily_cost_val, weekly_cost_val, daily_limit, weekly_limit
-    )
+    usage_display = format_usage_percentages(five_hour_pct, seven_day_pct)
 
     # Detect terminal width for responsive layout
     term_width = get_terminal_width()
@@ -841,22 +1062,14 @@ def _row1_visible_len(d: str, b: str) -> int:
     elif term_width >= 60:
         # Compact: context bar | compact usage (drop cost, duration)
         compact_usage = format_usage_percentages(
-            daily_cost_val,
-            weekly_cost_val,
-            daily_limit,
-            weekly_limit,
-            compact=True,
+            five_hour_pct, seven_day_pct, compact=True
         )
         row2_parts = [context_info, compact_usage]
     else:
         # Minimal: just percentage + compact usage (drop bar, emoji, cost, duration)
         # Extract percentage from context_info by recalculating
         compact_usage = format_usage_percentages(
-            daily_cost_val,
-            weekly_cost_val,
-            daily_limit,
-            weekly_limit,
-            compact=True,
+            five_hour_pct, seven_day_pct, compact=True
         )
         # Build a minimal context display (just percentage, no bar)
         session_id = input_data.get("session_id", "")

From b6dfa301a615190c140c3adde645bb175e926aa0 Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 13:46:11 +0300
Subject: [PATCH 07/13] fix: support multiple OAuth token sources for usage API

---
 scripts/statusline.py | 237 +++++++++++++++++++++++++++++++++---------
 1 file changed, 186 insertions(+), 51 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 2ff0ab0..a34ff67 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -10,6 +10,7 @@
 
 import io
 import json
+import logging
 import os
 import subprocess
 import sys
@@ -17,6 +18,9 @@
 import time
 from datetime import datetime, timedelta
 from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
 
 # Cache configuration
 COST_CACHE_TTL_SECONDS = 300  # Cache costs for 5 minutes (cost data changes slowly)
@@ -607,36 +611,125 @@ def _is_usage_refresh_locked() -> bool:
     return False
 
 
-def fetch_usage_data() -> tuple[float | None, float | None]:
-    """Fetch usage data from Anthropic OAuth usage API.
+def get_oauth_token() -> str | None:
+    """Find OAuth token from multiple sources.
 
-    Reads OAuth token from ~/.claude/.credentials.json and calls the API.
+    Checks in order:
+    1. CLAUDE_CODE_OAUTH_TOKEN env var (most common for CLI sessions)
+    2. ~/.claude/.credentials.json -> .claudeAiOauth.accessToken (macOS/Linux)
+    3. %APPDATA%/claude/.credentials.json -> .claudeAiOauth.accessToken (Windows)
 
     Returns:
-        Tuple of (five_hour_pct, seven_day_pct) or (None, None) on failure.
+        OAuth access token string, or None if not found.
     """
-    import urllib.request
-    import urllib.error
+    # Source 1: Environment variable
+    env_token = os.environ.get("CLAUDE_CODE_OAUTH_TOKEN")
+    if env_token:
+        debug_log("OAuth token found via CLAUDE_CODE_OAUTH_TOKEN env var")
+        return env_token
 
-    # Read OAuth token
+    # Source 2: macOS/Linux credentials file
     creds_path = Path.home() / ".claude" / ".credentials.json"
-    if not creds_path.exists():
-        # Windows fallback
-        appdata = os.environ.get("APPDATA", "")
-        if appdata:
-            creds_path = Path(appdata) / "claude" / ".credentials.json"
-    if not creds_path.exists():
-        debug_log("No credentials file found for usage API")
-        return None, None
+    token = _read_token_from_creds(creds_path)
+    if token:
+        debug_log("OAuth token found via ~/.claude/.credentials.json")
+        return token
+
+    # Source 3: Windows credentials file
+    appdata = os.environ.get("APPDATA", "")
+    if appdata:
+        win_creds_path = Path(appdata) / "claude" / ".credentials.json"
+        token = _read_token_from_creds(win_creds_path)
+        if token:
+            debug_log("OAuth token found via APPDATA credentials")
+            return token
+
+    debug_log("No OAuth token found from any source")
+    return None
+
 
+def _read_token_from_creds(creds_path: Path) -> str | None:
+    """Extract OAuth access token from a credentials JSON file.
+
+    Args:
+        creds_path: Path to the .credentials.json file.
+
+    Returns:
+        Token string, or None if file missing/invalid/no token.
+    """
+    if not creds_path.exists():
+        return None
     try:
         creds = json.loads(creds_path.read_text(encoding="utf-8"))
-        token = creds.get("claudeAiOauth", {}).get("accessToken")
-        if not token:
-            debug_log("No OAuth access token found in credentials")
-            return None, None
+        return creds.get("claudeAiOauth", {}).get("accessToken") or None
     except (json.JSONDecodeError, OSError) as e:
-        debug_log(f"Failed to read credentials: {e}")
+        debug_log(f"Failed to read credentials from {creds_path}: {e}")
+        return None
+
+
+def _urlopen_with_ssl_fallback(req: Any, timeout: int = 10) -> bytes:
+    """Open a URL request with SSL certificate error fallback.
+
+    Tries in order:
+    1. Default SSL context
+    2. Explicit /etc/ssl/cert.pem context (macOS fallback)
+    3. Unverified SSL as last resort (with warning)
+
+    Args:
+        req: The urllib.request.Request object.
+        timeout: Request timeout in seconds.
+
+    Returns:
+        Response body bytes.
+    """
+    import ssl
+    import urllib.error
+    import urllib.request
+
+    # Attempt 1: default SSL
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310
+            return resp.read()
+    except urllib.error.URLError as e:
+        if not isinstance(getattr(e, "reason", None), ssl.SSLError):
+            raise
+        debug_log(f"SSL error with default context: {e.reason}")
+
+    # Attempt 2: explicit cert bundle (common macOS path)
+    cert_pem = Path("/etc/ssl/cert.pem")
+    if cert_pem.exists():
+        try:
+            ctx = ssl.create_default_context(cafile=str(cert_pem))
+            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:  # noqa: S310
+                return resp.read()
+        except urllib.error.URLError as e:
+            if not isinstance(getattr(e, "reason", None), ssl.SSLError):
+                raise
+            debug_log(f"SSL error with /etc/ssl/cert.pem: {e.reason}")
+
+    # Attempt 3: unverified SSL (last resort)
+    logger.warning("Falling back to unverified SSL for usage API request")
+    debug_log("WARNING: using unverified SSL context as last resort")
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:  # noqa: S310
+        return resp.read()
+
+
+def fetch_usage_data() -> tuple[float | None, float | None]:
+    """Fetch usage data from Anthropic OAuth usage API.
+
+    Reads OAuth token from multiple sources and calls the API.
+
+    Returns:
+        Tuple of (five_hour_pct, seven_day_pct) or (None, None) on failure.
+    """
+    import urllib.error
+    import urllib.request
+
+    token = get_oauth_token()
+    if not token:
         return None, None
 
     # Call the API
@@ -649,8 +742,8 @@ def fetch_usage_data() -> tuple[float | None, float | None]:
         },
     )
     try:
-        with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
-            data = json.loads(resp.read().decode("utf-8"))
+        body = _urlopen_with_ssl_fallback(req, timeout=10)
+        data = json.loads(body.decode("utf-8"))
         five_hour = data.get("five_hour", {}).get("utilization")
         seven_day = data.get("seven_day", {}).get("utilization")
         debug_log(f"Usage API: 5h={five_hour}%, 7d={seven_day}%")
@@ -676,7 +769,7 @@ def spawn_usage_background_refresh() -> None:
     debug_log("Spawning usage background refresh process")
 
     refresh_script = f"""
-import json, time, sys, os
+import json, time, sys, os, ssl, logging
 import urllib.request
 import urllib.error
 from pathlib import Path
@@ -685,41 +778,83 @@ def spawn_usage_background_refresh() -> None:
 cache_file = Path({str(USAGE_CACHE_FILE)!r})
 BACKOFF_SECONDS = {USAGE_429_BACKOFF_SECONDS}
 
+def get_token():
+    # Source 1: env var
+    t = os.environ.get("CLAUDE_CODE_OAUTH_TOKEN")
+    if t:
+        return t
+    # Source 2: macOS/Linux creds
+    cp = Path.home() / ".claude" / ".credentials.json"
+    t = _read(cp)
+    if t:
+        return t
+    # Source 3: Windows creds
+    ad = os.environ.get("APPDATA", "")
+    if ad:
+        t = _read(Path(ad) / "claude" / ".credentials.json")
+        if t:
+            return t
+    return None
+
+def _read(p):
+    if not p.exists():
+        return None
+    try:
+        c = json.loads(p.read_text(encoding="utf-8"))
+        return c.get("claudeAiOauth", {{}}).get("accessToken") or None
+    except Exception:
+        return None
+
+def urlopen_ssl(req, timeout=10):
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as r:
+            return r.read()
+    except urllib.error.URLError as e:
+        if not isinstance(getattr(e, "reason", None), ssl.SSLError):
+            raise
+    cp = Path("/etc/ssl/cert.pem")
+    if cp.exists():
+        try:
+            ctx = ssl.create_default_context(cafile=str(cp))
+            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
+                return r.read()
+        except urllib.error.URLError as e:
+            if not isinstance(getattr(e, "reason", None), ssl.SSLError):
+                raise
+    logging.warning("Falling back to unverified SSL")
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+    with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
+        return r.read()
+
 try:
     lock_file.write_text(str(time.time()))
 
-    creds_path = Path.home() / ".claude" / ".credentials.json"
-    if not creds_path.exists():
-        appdata = os.environ.get("APPDATA", "")
-        if appdata:
-            creds_path = Path(appdata) / "claude" / ".credentials.json"
-
     five_hour_pct = None
     seven_day_pct = None
     error_429 = False
 
-    if creds_path.exists():
-        creds = json.loads(creds_path.read_text(encoding="utf-8"))
-        token = creds.get("claudeAiOauth", {{}}).get("accessToken")
-        if token:
-            url = "https://api.anthropic.com/api/oauth/usage"
-            req = urllib.request.Request(
-                url,
-                headers={{
-                    "Authorization": f"Bearer {{token}}",
-                    "anthropic-beta": "oauth-2025-04-20",
-                }},
-            )
-            try:
-                with urllib.request.urlopen(req, timeout=10) as resp:
-                    data = json.loads(resp.read().decode("utf-8"))
-                five_hour_pct = data.get("five_hour", {{}}).get("utilization")
-                seven_day_pct = data.get("seven_day", {{}}).get("utilization")
-            except urllib.error.HTTPError as e:
-                if e.code == 429:
-                    error_429 = True
-            except Exception:
-                pass
+    token = get_token()
+    if token:
+        url = "https://api.anthropic.com/api/oauth/usage"
+        req = urllib.request.Request(
+            url,
+            headers={{
+                "Authorization": f"Bearer {{token}}",
+                "anthropic-beta": "oauth-2025-04-20",
+            }},
+        )
+        try:
+            body = urlopen_ssl(req, timeout=10)
+            data = json.loads(body.decode("utf-8"))
+            five_hour_pct = data.get("five_hour", {{}}).get("utilization")
+            seven_day_pct = data.get("seven_day", {{}}).get("utilization")
+        except urllib.error.HTTPError as e:
+            if e.code == 429:
+                error_429 = True
+        except Exception:
+            pass
 
     cache = {{
         "five_hour_pct": five_hour_pct,

From c111c192cbf9a858c9cdc5dc19378e623fbd846c Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 14:40:09 +0300
Subject: [PATCH 08/13] fix: respect Retry-After header from usage API 429
 responses

---
 scripts/statusline.py | 27 ++++++++++++++++-----------
 1 file changed, 16 insertions(+), 11 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index a34ff67..e360ec2 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -31,7 +31,9 @@
 USAGE_CACHE_TTL_SECONDS = 120  # 2 minutes — aggressive but avoids 429s
 USAGE_CACHE_FILE = Path(tempfile.gettempdir()) / "statusline_usage_cache.json"
 USAGE_REFRESH_LOCK = Path(tempfile.gettempdir()) / "statusline_usage_refresh.lock"  # noqa: S108
-USAGE_429_BACKOFF_SECONDS = 300  # 5 minutes on rate limit
+USAGE_429_BACKOFF_SECONDS_DEFAULT = (
+    300  # 5 minutes on rate limit (fallback if no Retry-After)
+)
 
 # Force UTF-8 output on Windows (fixes emoji encoding errors)
 if sys.platform == "win32":
@@ -569,7 +571,7 @@ def load_usage_cache() -> dict:
 def save_usage_cache(
     five_hour_pct: float | None,
     seven_day_pct: float | None,
-    error_429: bool = False,
+    backoff_seconds: int = 0,
 ) -> None:
     """Save usage API values to cache file."""
     cache: dict = {
@@ -577,8 +579,8 @@ def save_usage_cache(
         "seven_day_pct": seven_day_pct,
         "timestamp": time.time(),
     }
-    if error_429:
-        cache["backoff_until"] = time.time() + USAGE_429_BACKOFF_SECONDS
+    if backoff_seconds > 0:
+        cache["backoff_until"] = time.time() + backoff_seconds
     try:
         USAGE_CACHE_FILE.write_text(json.dumps(cache), encoding="utf-8")
     except OSError:
@@ -750,8 +752,11 @@ def fetch_usage_data() -> tuple[float | None, float | None]:
         return five_hour, seven_day
     except urllib.error.HTTPError as e:
         if e.code == 429:
-            debug_log("Usage API rate limited (429), backing off 5 minutes")
-            save_usage_cache(None, None, error_429=True)
+            retry_after = int(
+                e.headers.get("Retry-After", USAGE_429_BACKOFF_SECONDS_DEFAULT)
+            )
+            debug_log(f"Usage API rate limited (429), backing off {retry_after}s")
+            save_usage_cache(None, None, backoff_seconds=retry_after)
         else:
             debug_log(f"Usage API HTTP error: {e.code}")
         return None, None
@@ -776,7 +781,7 @@ def spawn_usage_background_refresh() -> None:
 
 lock_file = Path({str(USAGE_REFRESH_LOCK)!r})
 cache_file = Path({str(USAGE_CACHE_FILE)!r})
-BACKOFF_SECONDS = {USAGE_429_BACKOFF_SECONDS}
+BACKOFF_SECONDS_DEFAULT = {USAGE_429_BACKOFF_SECONDS_DEFAULT}
 
 def get_token():
     # Source 1: env var
@@ -833,7 +838,7 @@ def urlopen_ssl(req, timeout=10):
 
     five_hour_pct = None
     seven_day_pct = None
-    error_429 = False
+    backoff_seconds = 0
 
     token = get_token()
     if token:
@@ -852,7 +857,7 @@ def urlopen_ssl(req, timeout=10):
             seven_day_pct = data.get("seven_day", {{}}).get("utilization")
         except urllib.error.HTTPError as e:
             if e.code == 429:
-                error_429 = True
+                backoff_seconds = int(e.headers.get("Retry-After", BACKOFF_SECONDS_DEFAULT))
         except Exception:
             pass
 
@@ -861,8 +866,8 @@ def urlopen_ssl(req, timeout=10):
         "seven_day_pct": seven_day_pct,
         "timestamp": time.time(),
     }}
-    if error_429:
-        cache["backoff_until"] = time.time() + BACKOFF_SECONDS
+    if backoff_seconds > 0:
+        cache["backoff_until"] = time.time() + backoff_seconds
     cache_file.write_text(json.dumps(cache))
 finally:
     lock_file.unlink(missing_ok=True)

From 00931f57a9d9010d3de598e1776e7ec68ef606ef Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 15:07:02 +0300
Subject: [PATCH 09/13] refactor: simplify usage API to inline fetch with
 cache-on-failure pattern

---
 scripts/statusline.py | 304 ++++++++++--------------------------------
 1 file changed, 71 insertions(+), 233 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index e360ec2..4aac302 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -28,12 +28,8 @@
 COST_REFRESH_LOCK = Path(tempfile.gettempdir()) / "statusline_cost_refresh.lock"  # noqa: S108
 
 # Usage API cache (separate from cost cache — different TTL)
-USAGE_CACHE_TTL_SECONDS = 120  # 2 minutes — aggressive but avoids 429s
+USAGE_CACHE_TTL_SECONDS = 60  # 60 second TTL
 USAGE_CACHE_FILE = Path(tempfile.gettempdir()) / "statusline_usage_cache.json"
-USAGE_REFRESH_LOCK = Path(tempfile.gettempdir()) / "statusline_usage_refresh.lock"  # noqa: S108
-USAGE_429_BACKOFF_SECONDS_DEFAULT = (
-    300  # 5 minutes on rate limit (fallback if no Retry-After)
-)
 
 # Force UTF-8 output on Windows (fixes emoji encoding errors)
 if sys.platform == "win32":
@@ -558,59 +554,39 @@ def spawn_background_refresh(cwd: str) -> None:
         debug_log(f"Failed to spawn background refresh: {e}")
 
 
-def load_usage_cache() -> dict:
-    """Load usage API cache from file."""
-    try:
-        if USAGE_CACHE_FILE.exists():
-            return json.loads(USAGE_CACHE_FILE.read_text(encoding="utf-8"))
-    except (json.JSONDecodeError, OSError):
-        pass
-    return {}
-
-
-def save_usage_cache(
-    five_hour_pct: float | None,
-    seven_day_pct: float | None,
-    backoff_seconds: int = 0,
-) -> None:
-    """Save usage API values to cache file."""
-    cache: dict = {
-        "five_hour_pct": five_hour_pct,
-        "seven_day_pct": seven_day_pct,
-        "timestamp": time.time(),
-    }
-    if backoff_seconds > 0:
-        cache["backoff_until"] = time.time() + backoff_seconds
-    try:
-        USAGE_CACHE_FILE.write_text(json.dumps(cache), encoding="utf-8")
-    except OSError:
-        pass
+def fetch_usage_data() -> tuple[float | None, float | None]:
+    """Fetch usage data from Anthropic OAuth usage API.
 
+    Simple inline fetch with SSL fallback. On ANY failure, returns (None, None).
 
-def is_usage_cache_valid(cache: dict) -> bool:
-    """Check if usage cache is still valid."""
-    if not cache:
-        return False
-    # If in 429 backoff, cache is valid until backoff expires
-    backoff_until = cache.get("backoff_until", 0)
-    if backoff_until and time.time() < backoff_until:
-        return True
-    cache_time = cache.get("timestamp", 0)
-    age = time.time() - cache_time
-    return age < USAGE_CACHE_TTL_SECONDS
+    Returns:
+        Tuple of (five_hour_pct, seven_day_pct) or (None, None) on failure.
+    """
+    import urllib.request
 
+    token = get_oauth_token()
+    if not token:
+        return None, None
 
-def _is_usage_refresh_locked() -> bool:
-    """Check if a usage background refresh is already running."""
+    url = "https://api.anthropic.com/api/oauth/usage"
+    req = urllib.request.Request(  # noqa: S310
+        url,
+        headers={
+            "Authorization": f"Bearer {token}",
+            "anthropic-beta": "oauth-2025-04-20",
+            "User-Agent": "claude-code/1.0",
+        },
+    )
     try:
-        if USAGE_REFRESH_LOCK.exists():
-            lock_age = time.time() - USAGE_REFRESH_LOCK.stat().st_mtime
-            if lock_age < 60:
-                return True
-            USAGE_REFRESH_LOCK.unlink(missing_ok=True)
-    except OSError:
-        pass
-    return False
+        body = _urlopen_with_ssl_fallback(req, timeout=5)
+        data = json.loads(body.decode("utf-8"))
+        five_hour = data.get("five_hour", {}).get("utilization")
+        seven_day = data.get("seven_day", {}).get("utilization")
+        debug_log(f"Usage API: 5h={five_hour}%, 7d={seven_day}%")
+        return five_hour, seven_day
+    except Exception as e:
+        debug_log(f"Usage API error: {e}")
+        return None, None
 
 
 def get_oauth_token() -> str | None:
@@ -719,198 +695,60 @@ def _urlopen_with_ssl_fallback(req: Any, timeout: int = 10) -> bytes:
         return resp.read()
 
 
-def fetch_usage_data() -> tuple[float | None, float | None]:
-    """Fetch usage data from Anthropic OAuth usage API.
+def get_usage_cached() -> tuple[float | None, float | None]:
+    """Get usage percentages with simple inline cache.
 
-    Reads OAuth token from multiple sources and calls the API.
+    If cache is fresh (< 60s): return cached values.
+    If cache is stale: fetch inline, update cache on success.
+    If fetch fails and stale cache exists: re-save stale data with fresh
+    timestamp (implicit backoff — won't retry for another 60s).
 
     Returns:
-        Tuple of (five_hour_pct, seven_day_pct) or (None, None) on failure.
+        Tuple of (five_hour_pct, seven_day_pct) — None means data unavailable.
     """
-    import urllib.error
-    import urllib.request
-
-    token = get_oauth_token()
-    if not token:
-        return None, None
-
-    # Call the API
-    url = "https://api.anthropic.com/api/oauth/usage"
-    req = urllib.request.Request(  # noqa: S310
-        url,
-        headers={
-            "Authorization": f"Bearer {token}",
-            "anthropic-beta": "oauth-2025-04-20",
-        },
-    )
+    # Try to load existing cache
+    cache: dict = {}
     try:
-        body = _urlopen_with_ssl_fallback(req, timeout=10)
-        data = json.loads(body.decode("utf-8"))
-        five_hour = data.get("five_hour", {}).get("utilization")
-        seven_day = data.get("seven_day", {}).get("utilization")
-        debug_log(f"Usage API: 5h={five_hour}%, 7d={seven_day}%")
-        return five_hour, seven_day
-    except urllib.error.HTTPError as e:
-        if e.code == 429:
-            retry_after = int(
-                e.headers.get("Retry-After", USAGE_429_BACKOFF_SECONDS_DEFAULT)
-            )
-            debug_log(f"Usage API rate limited (429), backing off {retry_after}s")
-            save_usage_cache(None, None, backoff_seconds=retry_after)
-        else:
-            debug_log(f"Usage API HTTP error: {e.code}")
-        return None, None
-    except (urllib.error.URLError, OSError, json.JSONDecodeError) as e:
-        debug_log(f"Usage API error: {e}")
-        return None, None
-
-
-def spawn_usage_background_refresh() -> None:
-    """Spawn a background process to refresh the usage API cache."""
-    if _is_usage_refresh_locked():
-        debug_log("Usage background refresh already running, skipping")
-        return
-
-    debug_log("Spawning usage background refresh process")
-
-    refresh_script = f"""
-import json, time, sys, os, ssl, logging
-import urllib.request
-import urllib.error
-from pathlib import Path
-
-lock_file = Path({str(USAGE_REFRESH_LOCK)!r})
-cache_file = Path({str(USAGE_CACHE_FILE)!r})
-BACKOFF_SECONDS_DEFAULT = {USAGE_429_BACKOFF_SECONDS_DEFAULT}
-
-def get_token():
-    # Source 1: env var
-    t = os.environ.get("CLAUDE_CODE_OAUTH_TOKEN")
-    if t:
-        return t
-    # Source 2: macOS/Linux creds
-    cp = Path.home() / ".claude" / ".credentials.json"
-    t = _read(cp)
-    if t:
-        return t
-    # Source 3: Windows creds
-    ad = os.environ.get("APPDATA", "")
-    if ad:
-        t = _read(Path(ad) / "claude" / ".credentials.json")
-        if t:
-            return t
-    return None
-
-def _read(p):
-    if not p.exists():
-        return None
-    try:
-        c = json.loads(p.read_text(encoding="utf-8"))
-        return c.get("claudeAiOauth", {{}}).get("accessToken") or None
-    except Exception:
-        return None
+        if USAGE_CACHE_FILE.exists():
+            cache = json.loads(USAGE_CACHE_FILE.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        pass
 
-def urlopen_ssl(req, timeout=10):
-    try:
-        with urllib.request.urlopen(req, timeout=timeout) as r:
-            return r.read()
-    except urllib.error.URLError as e:
-        if not isinstance(getattr(e, "reason", None), ssl.SSLError):
-            raise
-    cp = Path("/etc/ssl/cert.pem")
-    if cp.exists():
+    # If cache is fresh, return it
+    if cache:
+        age = time.time() - cache.get("timestamp", 0)
+        if age < USAGE_CACHE_TTL_SECONDS:
+            debug_log(f"Using cached usage (age: {age:.1f}s)")
+            return cache.get("five_hour_pct"), cache.get("seven_day_pct")
+
+    # Cache stale or missing — fetch inline
+    five_h, seven_d = fetch_usage_data()
+
+    if five_h is not None or seven_d is not None:
+        # Fetch succeeded — save fresh data
+        new_cache = {
+            "five_hour_pct": five_h,
+            "seven_day_pct": seven_d,
+            "timestamp": time.time(),
+        }
         try:
-            ctx = ssl.create_default_context(cafile=str(cp))
-            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
-                return r.read()
-        except urllib.error.URLError as e:
-            if not isinstance(getattr(e, "reason", None), ssl.SSLError):
-                raise
-    logging.warning("Falling back to unverified SSL")
-    ctx = ssl.create_default_context()
-    ctx.check_hostname = False
-    ctx.verify_mode = ssl.CERT_NONE
-    with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
-        return r.read()
-
-try:
-    lock_file.write_text(str(time.time()))
-
-    five_hour_pct = None
-    seven_day_pct = None
-    backoff_seconds = 0
+            USAGE_CACHE_FILE.write_text(json.dumps(new_cache), encoding="utf-8")
+        except OSError:
+            pass
+        return five_h, seven_d
 
-    token = get_token()
-    if token:
-        url = "https://api.anthropic.com/api/oauth/usage"
-        req = urllib.request.Request(
-            url,
-            headers={{
-                "Authorization": f"Bearer {{token}}",
-                "anthropic-beta": "oauth-2025-04-20",
-            }},
-        )
+    # Fetch failed — if stale cache exists, re-save with fresh timestamp
+    # (implicit backoff: won't retry for another USAGE_CACHE_TTL_SECONDS)
+    if cache:
+        cache["timestamp"] = time.time()
         try:
-            body = urlopen_ssl(req, timeout=10)
-            data = json.loads(body.decode("utf-8"))
-            five_hour_pct = data.get("five_hour", {{}}).get("utilization")
-            seven_day_pct = data.get("seven_day", {{}}).get("utilization")
-        except urllib.error.HTTPError as e:
-            if e.code == 429:
-                backoff_seconds = int(e.headers.get("Retry-After", BACKOFF_SECONDS_DEFAULT))
-        except Exception:
+            USAGE_CACHE_FILE.write_text(json.dumps(cache), encoding="utf-8")
+        except OSError:
             pass
-
-    cache = {{
-        "five_hour_pct": five_hour_pct,
-        "seven_day_pct": seven_day_pct,
-        "timestamp": time.time(),
-    }}
-    if backoff_seconds > 0:
-        cache["backoff_until"] = time.time() + backoff_seconds
-    cache_file.write_text(json.dumps(cache))
-finally:
-    lock_file.unlink(missing_ok=True)
-"""
-
-    try:
-        subprocess.Popen(  # noqa: S603
-            [sys.executable, "-c", refresh_script],  # noqa: S603
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            start_new_session=True,
-        )
-    except OSError as e:
-        debug_log(f"Failed to spawn usage background refresh: {e}")
-
-
-def get_usage_cached() -> tuple[float | None, float | None]:
-    """Get usage percentages with non-blocking cache refresh.
-
-    Returns cached values immediately. If cache is expired, returns stale
-    values and spawns a background refresh.
-
-    Returns:
-        Tuple of (five_hour_pct, seven_day_pct) — None means data unavailable.
-    """
-    cache = load_usage_cache()
-
-    if is_usage_cache_valid(cache):
-        debug_log(
-            f"Using cached usage (age: {time.time() - cache.get('timestamp', 0):.1f}s)"
-        )
+        debug_log("Fetch failed, re-stamped stale cache for implicit backoff")
         return cache.get("five_hour_pct"), cache.get("seven_day_pct")
 
-    # Cache stale or missing — return stale values and refresh in background
-    stale_5h = cache.get("five_hour_pct") if cache else None
-    stale_7d = cache.get("seven_day_pct") if cache else None
-    debug_log(
-        f"Usage cache expired/missing, stale values: 5h={stale_5h}, 7d={stale_7d}"
-    )
-
-    spawn_usage_background_refresh()
-
-    return stale_5h, stale_7d
+    return None, None
 
 
 def get_costs_cached(cwd: str) -> tuple[str, str, float, float]:

From b6d4f5d231c3b27a6a58c97257694d1adccfd59f Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 22:32:42 +0300
Subject: [PATCH 10/13] refactor: read usage percentages from statusLine JSON
 stdin instead of OAuth API

---
 scripts/statusline.py | 213 ++----------------------------------------
 1 file changed, 6 insertions(+), 207 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 4aac302..72b6fc6 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -10,7 +10,6 @@
 
 import io
 import json
-import logging
 import os
 import subprocess
 import sys
@@ -18,19 +17,12 @@
 import time
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Any
-
-logger = logging.getLogger(__name__)
 
 # Cache configuration
 COST_CACHE_TTL_SECONDS = 300  # Cache costs for 5 minutes (cost data changes slowly)
 COST_CACHE_FILE = Path(tempfile.gettempdir()) / "statusline_cost_cache.json"
 COST_REFRESH_LOCK = Path(tempfile.gettempdir()) / "statusline_cost_refresh.lock"  # noqa: S108
 
-# Usage API cache (separate from cost cache — different TTL)
-USAGE_CACHE_TTL_SECONDS = 60  # 60 second TTL
-USAGE_CACHE_FILE = Path(tempfile.gettempdir()) / "statusline_usage_cache.json"
-
 # Force UTF-8 output on Windows (fixes emoji encoding errors)
 if sys.platform == "win32":
     sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
@@ -554,203 +546,6 @@ def spawn_background_refresh(cwd: str) -> None:
         debug_log(f"Failed to spawn background refresh: {e}")
 
 
-def fetch_usage_data() -> tuple[float | None, float | None]:
-    """Fetch usage data from Anthropic OAuth usage API.
-
-    Simple inline fetch with SSL fallback. On ANY failure, returns (None, None).
-
-    Returns:
-        Tuple of (five_hour_pct, seven_day_pct) or (None, None) on failure.
-    """
-    import urllib.request
-
-    token = get_oauth_token()
-    if not token:
-        return None, None
-
-    url = "https://api.anthropic.com/api/oauth/usage"
-    req = urllib.request.Request(  # noqa: S310
-        url,
-        headers={
-            "Authorization": f"Bearer {token}",
-            "anthropic-beta": "oauth-2025-04-20",
-            "User-Agent": "claude-code/1.0",
-        },
-    )
-    try:
-        body = _urlopen_with_ssl_fallback(req, timeout=5)
-        data = json.loads(body.decode("utf-8"))
-        five_hour = data.get("five_hour", {}).get("utilization")
-        seven_day = data.get("seven_day", {}).get("utilization")
-        debug_log(f"Usage API: 5h={five_hour}%, 7d={seven_day}%")
-        return five_hour, seven_day
-    except Exception as e:
-        debug_log(f"Usage API error: {e}")
-        return None, None
-
-
-def get_oauth_token() -> str | None:
-    """Find OAuth token from multiple sources.
-
-    Checks in order:
-    1. CLAUDE_CODE_OAUTH_TOKEN env var (most common for CLI sessions)
-    2. ~/.claude/.credentials.json -> .claudeAiOauth.accessToken (macOS/Linux)
-    3. %APPDATA%/claude/.credentials.json -> .claudeAiOauth.accessToken (Windows)
-
-    Returns:
-        OAuth access token string, or None if not found.
-    """
-    # Source 1: Environment variable
-    env_token = os.environ.get("CLAUDE_CODE_OAUTH_TOKEN")
-    if env_token:
-        debug_log("OAuth token found via CLAUDE_CODE_OAUTH_TOKEN env var")
-        return env_token
-
-    # Source 2: macOS/Linux credentials file
-    creds_path = Path.home() / ".claude" / ".credentials.json"
-    token = _read_token_from_creds(creds_path)
-    if token:
-        debug_log("OAuth token found via ~/.claude/.credentials.json")
-        return token
-
-    # Source 3: Windows credentials file
-    appdata = os.environ.get("APPDATA", "")
-    if appdata:
-        win_creds_path = Path(appdata) / "claude" / ".credentials.json"
-        token = _read_token_from_creds(win_creds_path)
-        if token:
-            debug_log("OAuth token found via APPDATA credentials")
-            return token
-
-    debug_log("No OAuth token found from any source")
-    return None
-
-
-def _read_token_from_creds(creds_path: Path) -> str | None:
-    """Extract OAuth access token from a credentials JSON file.
-
-    Args:
-        creds_path: Path to the .credentials.json file.
-
-    Returns:
-        Token string, or None if file missing/invalid/no token.
-    """
-    if not creds_path.exists():
-        return None
-    try:
-        creds = json.loads(creds_path.read_text(encoding="utf-8"))
-        return creds.get("claudeAiOauth", {}).get("accessToken") or None
-    except (json.JSONDecodeError, OSError) as e:
-        debug_log(f"Failed to read credentials from {creds_path}: {e}")
-        return None
-
-
-def _urlopen_with_ssl_fallback(req: Any, timeout: int = 10) -> bytes:
-    """Open a URL request with SSL certificate error fallback.
-
-    Tries in order:
-    1. Default SSL context
-    2. Explicit /etc/ssl/cert.pem context (macOS fallback)
-    3. Unverified SSL as last resort (with warning)
-
-    Args:
-        req: The urllib.request.Request object.
-        timeout: Request timeout in seconds.
-
-    Returns:
-        Response body bytes.
-    """
-    import ssl
-    import urllib.error
-    import urllib.request
-
-    # Attempt 1: default SSL
-    try:
-        with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310
-            return resp.read()
-    except urllib.error.URLError as e:
-        if not isinstance(getattr(e, "reason", None), ssl.SSLError):
-            raise
-        debug_log(f"SSL error with default context: {e.reason}")
-
-    # Attempt 2: explicit cert bundle (common macOS path)
-    cert_pem = Path("/etc/ssl/cert.pem")
-    if cert_pem.exists():
-        try:
-            ctx = ssl.create_default_context(cafile=str(cert_pem))
-            with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:  # noqa: S310
-                return resp.read()
-        except urllib.error.URLError as e:
-            if not isinstance(getattr(e, "reason", None), ssl.SSLError):
-                raise
-            debug_log(f"SSL error with /etc/ssl/cert.pem: {e.reason}")
-
-    # Attempt 3: unverified SSL (last resort)
-    logger.warning("Falling back to unverified SSL for usage API request")
-    debug_log("WARNING: using unverified SSL context as last resort")
-    ctx = ssl.create_default_context()
-    ctx.check_hostname = False
-    ctx.verify_mode = ssl.CERT_NONE
-    with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:  # noqa: S310
-        return resp.read()
-
-
-def get_usage_cached() -> tuple[float | None, float | None]:
-    """Get usage percentages with simple inline cache.
-
-    If cache is fresh (< 60s): return cached values.
-    If cache is stale: fetch inline, update cache on success.
-    If fetch fails and stale cache exists: re-save stale data with fresh
-    timestamp (implicit backoff — won't retry for another 60s).
-
-    Returns:
-        Tuple of (five_hour_pct, seven_day_pct) — None means data unavailable.
-    """
-    # Try to load existing cache
-    cache: dict = {}
-    try:
-        if USAGE_CACHE_FILE.exists():
-            cache = json.loads(USAGE_CACHE_FILE.read_text(encoding="utf-8"))
-    except (json.JSONDecodeError, OSError):
-        pass
-
-    # If cache is fresh, return it
-    if cache:
-        age = time.time() - cache.get("timestamp", 0)
-        if age < USAGE_CACHE_TTL_SECONDS:
-            debug_log(f"Using cached usage (age: {age:.1f}s)")
-            return cache.get("five_hour_pct"), cache.get("seven_day_pct")
-
-    # Cache stale or missing — fetch inline
-    five_h, seven_d = fetch_usage_data()
-
-    if five_h is not None or seven_d is not None:
-        # Fetch succeeded — save fresh data
-        new_cache = {
-            "five_hour_pct": five_h,
-            "seven_day_pct": seven_d,
-            "timestamp": time.time(),
-        }
-        try:
-            USAGE_CACHE_FILE.write_text(json.dumps(new_cache), encoding="utf-8")
-        except OSError:
-            pass
-        return five_h, seven_d
-
-    # Fetch failed — if stale cache exists, re-save with fresh timestamp
-    # (implicit backoff: won't retry for another USAGE_CACHE_TTL_SECONDS)
-    if cache:
-        cache["timestamp"] = time.time()
-        try:
-            USAGE_CACHE_FILE.write_text(json.dumps(cache), encoding="utf-8")
-        except OSError:
-            pass
-        debug_log("Fetch failed, re-stamped stale cache for implicit backoff")
-        return cache.get("five_hour_pct"), cache.get("seven_day_pct")
-
-    return None, None
-
-
 def get_costs_cached(cwd: str) -> tuple[str, str, float, float]:
     """Get daily and session costs with non-blocking cache refresh.
 
@@ -974,8 +769,12 @@ def main() -> None:
     daily_amount = daily_cost
     cost_display = f"{GREEN}\U0001f4b0 {daily_amount}{RESET}"
 
-    # Get real usage percentages from Anthropic OAuth API
-    five_hour_pct, seven_day_pct = get_usage_cached()
+    # Extract rate limits from statusLine JSON input
+    rate_limits = input_data.get("rate_limits", {})
+    five_hour = rate_limits.get("five_hour", {})
+    seven_day = rate_limits.get("seven_day", {})
+    five_hour_pct = five_hour.get("used_percentage")  # float or None
+    seven_day_pct = seven_day.get("used_percentage")  # float or None
 
     usage_display = format_usage_percentages(five_hour_pct, seven_day_pct)
 

From b1e182ebe61baf34a46fd75703331c44ff7c8cf7 Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 23:05:46 +0300
Subject: [PATCH 11/13] fix: address Qodo review - width detection, double
 read, stale docstring

---
 scripts/statusline.py | 80 ++++++++++++-------------------------------
 1 file changed, 22 insertions(+), 58 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 72b6fc6..19235d4 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -126,8 +126,13 @@ def find_session_file(session_id: str, current_dir: str) -> Path | None:
     return None
 
 
-def calculate_context_usage(input_data: dict) -> str | None:
-    """Calculate actual context usage from session file."""
+def calculate_context_usage(input_data: dict) -> tuple[str | None, float]:
+    """Calculate actual context usage from session file.
+
+    Returns:
+        Tuple of (formatted_string, raw_pct) where raw_pct is the usage
+        percentage as a float (e.g. 57.1), or 0.0 if unavailable.
+    """
     session_id = input_data.get("session_id", "")
     current_dir = input_data.get("cwd") or input_data.get("workspace", {}).get(
         "current_dir", ""
@@ -136,7 +141,7 @@ def calculate_context_usage(input_data: dict) -> str | None:
     debug_log(f"Session ID: {session_id}, Current Dir: {current_dir}")
 
     if not session_id:
-        return None
+        return None, 0.0
 
     # Determine max context based on model
     model_name = input_data.get("model", {}).get("display_name") or input_data.get(
@@ -151,7 +156,7 @@ def calculate_context_usage(input_data: dict) -> str | None:
 
     session_file = find_session_file(session_id, current_dir)
     if not session_file:
-        return None
+        return None, 0.0
 
     debug_log(f"Processing session file: {session_file}")
 
@@ -204,12 +209,12 @@ def calculate_context_usage(input_data: dict) -> str | None:
             if total_input > 0:
                 usage_rate = total_input * 100 / max_context
                 progress_bar = create_progress_bar(usage_rate, total_input, max_context)
-                return f"🧠 {progress_bar}"
+                return f"🧠 {progress_bar}", usage_rate
 
     except OSError as e:
         debug_log(f"Error reading session file: {e}")
 
-    return None
+    return None, 0.0
 
 
 def shorten_cwd(full_path: str) -> str:
@@ -256,19 +261,15 @@ def truncate_str(s: str, max_len: int) -> str:
 def get_terminal_width() -> int:
     """Get terminal width, defaulting to 120 if detection fails.
 
-    In Claude Code's statusline context, shutil.get_terminal_size() often
-    returns unreasonably small values because it's not a real TTY.
-    We treat anything below 40 columns as unreliable and default to 120.
+    Only falls back to 120 when detection truly fails (returns 0 or raises
+    an exception). Small positive values are allowed through so narrow
+    terminals get compact layouts.
     """
     try:
         import shutil
 
         width = shutil.get_terminal_size(fallback=(120, 24)).columns
-        # In non-TTY contexts (like Claude Code statusline), the detected
-        # width can be 0 or very small. Use a conservative default instead.
-        if width < 40:
-            return 120
-        return width
+        return width if width > 0 else 120
     except Exception:
         return 120
 
@@ -650,11 +651,11 @@ def format_usage_percentages(
 ) -> str:
     """Format 5h and weekly usage as colored percentage strings.
 
-    Uses real utilization values from the Anthropic OAuth usage API.
+    Uses values from the statusLine JSON stdin ``rate_limits`` field.
 
     Args:
-        five_hour_pct: 5-hour utilization percentage (0-100) from API, or None if unavailable.
-        seven_day_pct: 7-day utilization percentage (0-100) from API, or None if unavailable.
+        five_hour_pct: 5-hour utilization percentage (0-100) from rate_limits, or None if unavailable.
+        seven_day_pct: 7-day utilization percentage (0-100) from rate_limits, or None if unavailable.
         compact: If True, use shorter labels (e.g., "5h 2%·w 60%").
 
     Returns:
@@ -747,7 +748,7 @@ def main() -> None:
     claude_version = get_claude_version()
 
     # Get context info
-    context_info = calculate_context_usage(input_data)
+    context_info, ctx_raw_pct = calculate_context_usage(input_data)
     if not context_info:
         # Fallback empty progress bar
         max_context = 200000
@@ -844,49 +845,12 @@ def _row1_visible_len(d: str, b: str) -> int:
         row2_parts = [context_info, compact_usage]
     else:
         # Minimal: just percentage + compact usage (drop bar, emoji, cost, duration)
-        # Extract percentage from context_info by recalculating
         compact_usage = format_usage_percentages(
             five_hour_pct, seven_day_pct, compact=True
         )
-        # Build a minimal context display (just percentage, no bar)
-        session_id = input_data.get("session_id", "")
-        max_ctx = 200000
-        if "1M" in raw_model:
-            max_ctx = 1_000_000
-        # Try to get actual percentage from session data
-        ctx_pct = "0%"
-        if session_id:
-            session_file = find_session_file(
-                session_id,
-                input_data.get("cwd")
-                or input_data.get("workspace", {}).get("current_dir", ""),
-            )
-            if session_file:
-                try:
-                    last_reset_line = 0
-                    lines_list: list[str] = []
-                    with open(session_file, "r", encoding="utf-8") as f:
-                        for i, line in enumerate(f, 1):
-                            lines_list.append(line)
-                            if '"/clear"' in line or '"/compact"' in line:
-                                last_reset_line = i
-                    for line in lines_list[last_reset_line:]:
-                        try:
-                            entry = json.loads(line)
-                            usage_data = entry.get("message", {}).get("usage", {})
-                            if usage_data:
-                                total_in = (
-                                    usage_data.get("input_tokens", 0)
-                                    + usage_data.get("cache_read_input_tokens", 0)
-                                    + usage_data.get("cache_creation_input_tokens", 0)
-                                )
-                                if total_in > 0:
-                                    ctx_pct = f"{total_in * 100 / max_ctx:.0f}%"
-                        except json.JSONDecodeError:
-                            continue
-                except OSError:
-                    pass
-        row2_parts = [f"{ctx_pct}", compact_usage]
+        # Reuse raw percentage from calculate_context_usage() (no re-read)
+        ctx_pct = f"{ctx_raw_pct:.0f}%"
+        row2_parts = [ctx_pct, compact_usage]
 
     sys.stdout.write(" | ".join(row2_parts) + "\n")
 

From 3597474e1c63d8df064d7d9999638f97ff888099 Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 23:19:21 +0300
Subject: [PATCH 12/13] fix: correct usage color threshold boundaries (>=50
 yellow, >=75 red)

---
 scripts/statusline.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/scripts/statusline.py b/scripts/statusline.py
index 19235d4..76fdaa0 100755
--- a/scripts/statusline.py
+++ b/scripts/statusline.py
@@ -664,9 +664,9 @@ def format_usage_percentages(
     reset = "\033[0m"
 
     def color_for_pct(pct: float) -> str:
-        if pct > 75:
+        if pct >= 75:
             return "\033[31m"  # Red
-        elif pct > 50:
+        elif pct >= 50:
             return "\033[33m"  # Yellow
         return "\033[32m"  # Green
 

From 0a501ff24efc21b971b3bd9a356490146043b8ea Mon Sep 17 00:00:00 2001
From: Nadav Barkai <barkain@gmail.com>
Date: Fri, 3 Apr 2026 23:43:24 +0300
Subject: [PATCH 13/13] chore: bump version to 1.15.0

---
 .claude-plugin/marketplace.json | 2 +-
 .claude-plugin/plugin.json      | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json
index 6d9775e..2dc5b36 100644
--- a/.claude-plugin/marketplace.json
+++ b/.claude-plugin/marketplace.json
@@ -8,7 +8,7 @@
       "name": "workflow-orchestrator",
       "source": "./",
       "description": "Delegation system with workflow orchestration, specialized agents, and parallel execution for Claude Code",
-      "version": "1.14.0",
+      "version": "1.15.0",
       "author": {
         "name": "Nadav Barkai"
       },
diff --git a/.claude-plugin/plugin.json b/.claude-plugin/plugin.json
index f50818f..aa5c3ed 100644
--- a/.claude-plugin/plugin.json
+++ b/.claude-plugin/plugin.json
@@ -1,6 +1,6 @@
 {
   "name": "workflow-orchestrator",
-  "version": "1.14.0",
+  "version": "1.15.0",
   "description": "Delegation system with workflow orchestration, specialized agents, and parallel execution for Claude Code",
   "author": {
     "name": "Nadav Barkai"