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" diff --git a/scripts/statusline.py b/scripts/statusline.py index fcc9fdc..76fdaa0 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 @@ -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 @@ -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: @@ -241,8 +246,36 @@ 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. + + 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 + return width if width > 0 else 120 + 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 +284,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: @@ -275,17 +305,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 +353,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 +405,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 +426,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 +473,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 +484,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 +509,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 +526,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 +547,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 +560,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 +568,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 +617,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 +637,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 +644,60 @@ def get_turn_duration() -> str | None: return None +def format_usage_percentages( + five_hour_pct: float | None, + seven_day_pct: float | None, + compact: bool = False, +) -> str: + """Format 5h and weekly usage as colored percentage strings. + + Uses values from the statusLine JSON stdin ``rate_limits`` field. + + Args: + 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: + Formatted string like "5h 6% ยท weekly 1%" 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 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_str = "\u2026%" + five_h_color = "\033[90m" # Gray for placeholder + + if seven_day_pct is not None: + weekly_str = f"{seven_day_pct:.0f}%" + weekly_color = color_for_pct(seven_day_pct) + else: + weekly_str = "\u2026%" + weekly_color = "\033[90m" # Gray for placeholder + + if compact: + return ( + f"{five_h_color}5h {five_h_str}{reset}" + f"\u00b7" + f"{weekly_color}w {weekly_str}{reset}" + ) + + return ( + f"{five_h_color}5h {five_h_str}{reset}" + f" \u00b7 " + f"{weekly_color}weekly {weekly_str}{reset}" + ) + + def main() -> None: """Main entry point.""" # Clear debug log at start @@ -678,13 +740,15 @@ 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() # 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 @@ -693,8 +757,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()) @@ -702,24 +766,92 @@ 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: daily cost only daily_amount = daily_cost - session_amount = session_cost - cost_display = f"{GREEN}๐Ÿ’ฐ {daily_amount} (๐ŸŽฏ {session_amount}){RESET}" + cost_display = f"{GREEN}\U0001f4b0 {daily_amount}{RESET}" + + # 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) + + # 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 + " โšก" + ) - # 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}" + # 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, turn duration - row2_parts = [cost_display, context_info] - if turn_duration: - row2_parts.append(f"{YELLOW}โฑ๏ธ {turn_duration}{RESET}") + + # --- Row 2: context bar | usage | cost | duration --- + # 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( + 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) + compact_usage = format_usage_percentages( + five_hour_pct, seven_day_pct, compact=True + ) + # 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")