From 15984a96e7201b6ea13ba7b29e924238858a285d Mon Sep 17 00:00:00 2001 From: JeremyDev87 Date: Sat, 4 Apr 2026 21:51:32 +0900 Subject: [PATCH] feat(plugin): use exact Claude Code stdin telemetry in status bar (#1325) Prefer exact telemetry from Claude Code stdin over estimates: - cost: stdin cost.total_cost_usd > estimate_cost() (show $ vs ~$) - duration: stdin cost.total_duration_ms > hud-state timestamp - agent: stdin agent.name > CODINGBUDDY_ACTIVE_AGENT env - model: show display_name when available - rate_limits: show 5h/7d usage when present - worktree: show WT name when present Closes #1325 --- .../hooks/codingbuddy-hud.py | 139 +++++++++-- packages/claude-code-plugin/tests/test_hud.py | 222 +++++++++++++++++- 2 files changed, 339 insertions(+), 22 deletions(-) diff --git a/packages/claude-code-plugin/hooks/codingbuddy-hud.py b/packages/claude-code-plugin/hooks/codingbuddy-hud.py index 0b23e676..26c61ce3 100644 --- a/packages/claude-code-plugin/hooks/codingbuddy-hud.py +++ b/packages/claude-code-plugin/hooks/codingbuddy-hud.py @@ -1,8 +1,14 @@ #!/usr/bin/env python3 -"""CodingBuddy statusLine script (#1088). +"""CodingBuddy statusLine script (#1088, #1325). Claude Code invokes this via settings.json statusLine.command. Reads session data from stdin JSON, outputs formatted status to stdout. + +Telemetry fallback order per field: + cost → stdin cost.total_cost_usd > estimate_cost() + duration → stdin cost.total_duration_ms > hud-state sessionStartTimestamp + agent → stdin agent.name > CODINGBUDDY_ACTIVE_AGENT env + model → stdin model.display_name > model.id """ import json import os @@ -103,6 +109,16 @@ def get_health(ctx_pct: float) -> str: return "\U0001f7e2" # 🟢 +def format_duration_ms(ms) -> str: + """Format milliseconds to duration like '12m' or '1h23m'.""" + total_minutes = int(ms / 60_000) + if total_minutes < 60: + return f"{total_minutes}m" + hours = total_minutes // 60 + minutes = total_minutes % 60 + return f"{hours}h{minutes:02d}m" + + def format_duration(start_timestamp: str) -> str: """Format ISO timestamp to duration like '12m' or '1h23m'.""" try: @@ -138,41 +154,124 @@ def read_state(state_file: str = DEFAULT_STATE_FILE) -> dict: return {} +def resolve_cost(stdin_data: dict, model_id: str, ctx_window: dict) -> tuple: + """Resolve cost: stdin exact > estimate. Returns (cost, is_exact).""" + exact = (stdin_data.get("cost") or {}).get("total_cost_usd") + if exact is not None: + return (float(exact), True) + return (estimate_cost(model_id, ctx_window), False) + + +def resolve_duration(stdin_data: dict, hud_state: dict) -> str: + """Resolve duration: stdin exact > hud-state timestamp > '0m'.""" + exact_ms = (stdin_data.get("cost") or {}).get("total_duration_ms") + if exact_ms is not None: + return format_duration_ms(exact_ms) + start_ts = hud_state.get("sessionStartTimestamp", "") + if start_ts: + return format_duration(start_ts) + return "0m" + + +def resolve_agent(stdin_data: dict, env_agent: str = "") -> str: + """Resolve agent: stdin > env var.""" + stdin_agent = (stdin_data.get("agent") or {}).get("name", "") + return stdin_agent or env_agent + + +def resolve_model_label(stdin_data: dict) -> tuple: + """Resolve model info. Returns (model_id, display_label).""" + model_info = stdin_data.get("model") or {} + model_id = model_info.get("id", "") + display_name = model_info.get("display_name", "") + return (model_id, display_name) + + +def format_rate_limits(stdin_data: dict) -> str: + """Format rate-limit info if present. Returns '' when absent.""" + rl = stdin_data.get("rate_limits") + if not rl: + return "" + parts = [] + five = rl.get("five_hour") + if five: + pct = five.get("used_percentage", 0) + parts.append(f"5h:{pct:.0f}%") + seven = rl.get("seven_day") + if seven: + pct = seven.get("used_percentage", 0) + parts.append(f"7d:{pct:.0f}%") + if not parts: + return "" + return "RL:" + ",".join(parts) + + +def format_worktree(stdin_data: dict) -> str: + """Format worktree name if present. Returns '' when absent.""" + wt = stdin_data.get("worktree") + if not wt: + return "" + name = wt.get("name", "") + return f"WT:{name}" if name else "" + + def format_status_line( stdin_data: dict, hud_state: dict, active_agent: str = "", ) -> str: - """Format the statusLine output.""" + """Format the statusLine output. + + Fallback order per field: + cost → stdin cost.total_cost_usd > estimate_cost() + duration → stdin cost.total_duration_ms > hud-state sessionStartTimestamp + agent → stdin agent.name > active_agent param + model → stdin model.display_name > model.id + """ version = hud_state.get("version", "") mode = hud_state.get("currentMode") mode_label = mode if mode else "Ready" - ctx_window = stdin_data.get("context_window", {}) + ctx_window = stdin_data.get("context_window") or {} ctx_pct = ctx_window.get("used_percentage", 0) or 0 health = get_health(ctx_pct) - start_ts = hud_state.get("sessionStartTimestamp", "") - duration = format_duration(start_ts) if start_ts else "0m" - - model_id = "" - model_info = stdin_data.get("model", {}) - if model_info: - model_id = model_info.get("id", "") - - cost = estimate_cost(model_id, ctx_window) + model_id, display_name = resolve_model_label(stdin_data) + cost, is_exact = resolve_cost(stdin_data, model_id, ctx_window) + duration = resolve_duration(stdin_data, hud_state) cache = compute_cache_hit_rate(ctx_window) + agent = resolve_agent(stdin_data, active_agent) + + cost_prefix = "$" if is_exact else "~$" ver_str = f" v{version}" if version else "" - line1 = ( - f"{BUDDY_FACE} CB{ver_str} | {mode_label} {health} | " - f"{duration} | ~${cost:.2f} | Cache:{cache:.0f}% | Ctx:{ctx_pct:.0f}%" - ) - if not active_agent: + segments = [ + f"{BUDDY_FACE} CB{ver_str}", + f"{mode_label} {health}", + duration, + f"{cost_prefix}{cost:.2f}", + f"Cache:{cache:.0f}%", + f"Ctx:{ctx_pct:.0f}%", + ] + + rl = format_rate_limits(stdin_data) + if rl: + segments.append(rl) + + wt = format_worktree(stdin_data) + if wt: + segments.append(wt) + + if display_name: + segments.append(display_name) + + line1 = " | ".join(segments) + + if not agent: return line1 - return f"{line1}\n\U0001f916 {active_agent}" + return f"{line1}\n\U0001f916 {agent}" def main(): @@ -183,9 +282,9 @@ def main(): state_file = os.environ.get("CODINGBUDDY_HUD_STATE_FILE", DEFAULT_STATE_FILE) hud_state = read_state(state_file) - active_agent = os.environ.get("CODINGBUDDY_ACTIVE_AGENT", "") + env_agent = os.environ.get("CODINGBUDDY_ACTIVE_AGENT", "") - output = format_status_line(stdin_data, hud_state, active_agent) + output = format_status_line(stdin_data, hud_state, env_agent) print(output) except Exception: print(f"{BUDDY_FACE} CodingBuddy") diff --git a/packages/claude-code-plugin/tests/test_hud.py b/packages/claude-code-plugin/tests/test_hud.py index ad0cb2e5..c4f1c040 100644 --- a/packages/claude-code-plugin/tests/test_hud.py +++ b/packages/claude-code-plugin/tests/test_hud.py @@ -166,6 +166,122 @@ def test_empty_string(self): assert hud.format_duration("") == "0m" +class TestFormatDurationMs: + def test_zero(self): + assert hud.format_duration_ms(0) == "0m" + + def test_minutes(self): + assert hud.format_duration_ms(720_000) == "12m" + + def test_hours(self): + assert hud.format_duration_ms(4_980_000) == "1h23m" + + def test_sub_minute(self): + assert hud.format_duration_ms(30_000) == "0m" + + +class TestResolveCost: + def test_exact_from_stdin(self): + stdin = {"cost": {"total_cost_usd": 1.23}} + cost, is_exact = hud.resolve_cost(stdin, "", {}) + assert cost == 1.23 + assert is_exact is True + + def test_fallback_to_estimate(self): + stdin = {} + ctx = {"current_usage": { + "input_tokens": 10000, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 0, + }} + cost, is_exact = hud.resolve_cost(stdin, "claude-sonnet-4-5", ctx) + assert cost > 0 + assert is_exact is False + + def test_zero_exact_cost_is_exact(self): + stdin = {"cost": {"total_cost_usd": 0}} + cost, is_exact = hud.resolve_cost(stdin, "", {}) + assert cost == 0.0 + assert is_exact is True + + def test_missing_cost_key(self): + stdin = {"cost": {}} + cost, is_exact = hud.resolve_cost(stdin, "", {}) + assert is_exact is False + + +class TestResolveDuration: + def test_exact_from_stdin(self): + stdin = {"cost": {"total_duration_ms": 720_000}} + assert hud.resolve_duration(stdin, {}) == "12m" + + def test_fallback_to_hud_state(self): + ts = (datetime.now(timezone.utc) - timedelta(minutes=5)).isoformat() + stdin = {} + state = {"sessionStartTimestamp": ts} + result = hud.resolve_duration(stdin, state) + assert "5m" in result or "4m" in result + + def test_no_data_returns_zero(self): + assert hud.resolve_duration({}, {}) == "0m" + + +class TestResolveAgent: + def test_stdin_agent_preferred(self): + stdin = {"agent": {"name": "security-reviewer"}} + assert hud.resolve_agent(stdin, "old-env-agent") == "security-reviewer" + + def test_fallback_to_env(self): + assert hud.resolve_agent({}, "env-agent") == "env-agent" + + def test_both_empty(self): + assert hud.resolve_agent({}, "") == "" + + +class TestResolveModelLabel: + def test_display_name_and_id(self): + stdin = {"model": {"id": "claude-opus-4-6", "display_name": "Opus"}} + mid, display = hud.resolve_model_label(stdin) + assert mid == "claude-opus-4-6" + assert display == "Opus" + + def test_no_model(self): + mid, display = hud.resolve_model_label({}) + assert mid == "" + assert display == "" + + +class TestFormatRateLimits: + def test_no_rate_limits(self): + assert hud.format_rate_limits({}) == "" + + def test_five_hour_only(self): + stdin = {"rate_limits": {"five_hour": {"used_percentage": 23.5}}} + assert hud.format_rate_limits(stdin) == "RL:5h:24%" + + def test_both_limits(self): + stdin = {"rate_limits": { + "five_hour": {"used_percentage": 10}, + "seven_day": {"used_percentage": 40}, + }} + result = hud.format_rate_limits(stdin) + assert "5h:10%" in result + assert "7d:40%" in result + + +class TestFormatWorktree: + def test_no_worktree(self): + assert hud.format_worktree({}) == "" + + def test_with_name(self): + stdin = {"worktree": {"name": "my-feature", "path": "/tmp/wt"}} + assert hud.format_worktree(stdin) == "WT:my-feature" + + def test_empty_name(self): + stdin = {"worktree": {"path": "/tmp/wt"}} + assert hud.format_worktree(stdin) == "" + + class TestFormatStatusLine: def test_full_output_with_mode(self): stdin = { @@ -189,8 +305,27 @@ def test_full_output_with_mode(self): assert "\u25d5\u203f\u25d5" in result # ◕‿◕ assert "PLAN" in result assert "5.1.1" in result - assert "$" in result + assert "~$" in result # estimated (no cost.total_cost_usd) assert "Ctx:45%" in result + assert "Opus" in result # display_name shown + + def test_exact_cost_prefix(self): + stdin = { + "cost": {"total_cost_usd": 0.42, "total_duration_ms": 60000}, + "model": {"id": "claude-sonnet-4-5"}, + "context_window": {"used_percentage": 10}, + } + result = hud.format_status_line(stdin, {}) + assert "$0.42" in result + assert "~$" not in result # exact, not estimated + + def test_exact_duration_from_stdin(self): + stdin = { + "cost": {"total_duration_ms": 720_000}, + "context_window": {"used_percentage": 0}, + } + result = hud.format_status_line(stdin, {}) + assert "12m" in result def test_no_mode_shows_ready(self): result = hud.format_status_line({}, {"version": "5.1.1"}) @@ -200,7 +335,14 @@ def test_empty_state(self): result = hud.format_status_line({}, {}) assert "\u25d5\u203f\u25d5" in result # always has buddy face - def test_agent_line(self): + def test_agent_from_stdin(self): + stdin = {"agent": {"name": "security-reviewer"}} + result = hud.format_status_line(stdin, {}) + lines = result.strip().split("\n") + assert len(lines) == 2 + assert "security-reviewer" in lines[1] + + def test_agent_line_env_fallback(self): result = hud.format_status_line( {}, {"version": "5.1.1", "currentMode": "ACT"}, @@ -210,10 +352,61 @@ def test_agent_line(self): assert len(lines) == 2 assert "architect" in lines[1] + def test_stdin_agent_overrides_env(self): + stdin = {"agent": {"name": "from-stdin"}} + result = hud.format_status_line(stdin, {}, active_agent="from-env") + assert "from-stdin" in result + assert "from-env" not in result + def test_no_agent_single_line(self): result = hud.format_status_line({}, {"version": "5.1.1"}) assert "\n" not in result + def test_rate_limits_shown(self): + stdin = {"rate_limits": { + "five_hour": {"used_percentage": 50}, + "seven_day": {"used_percentage": 20}, + }} + result = hud.format_status_line(stdin, {}) + assert "RL:" in result + assert "5h:50%" in result + + def test_worktree_shown(self): + stdin = {"worktree": {"name": "feat-x"}} + result = hud.format_status_line(stdin, {}) + assert "WT:feat-x" in result + + def test_full_telemetry(self): + """All exact stdin fields present — full telemetry line.""" + stdin = { + "model": {"id": "claude-opus-4-6", "display_name": "Opus"}, + "cost": {"total_cost_usd": 2.50, "total_duration_ms": 4_980_000}, + "context_window": { + "used_percentage": 65, + "current_usage": { + "input_tokens": 5000, + "cache_creation_input_tokens": 1000, + "cache_read_input_tokens": 3000, + }, + }, + "rate_limits": { + "five_hour": {"used_percentage": 30}, + "seven_day": {"used_percentage": 15}, + }, + "worktree": {"name": "wt-1"}, + "agent": {"name": "test-agent"}, + } + state = {"version": "5.3.0", "currentMode": "ACT"} + result = hud.format_status_line(stdin, state) + assert "$2.50" in result + assert "~$" not in result + assert "1h23m" in result + assert "ACT" in result + assert "Opus" in result + assert "RL:" in result + assert "WT:wt-1" in result + assert "test-agent" in result + class TestIntegration: def test_pipe_stdin(self): @@ -223,6 +416,7 @@ def test_pipe_stdin(self): "transcript_path": "/tmp/t", "cwd": "/tmp", "model": {"id": "claude-opus-4-6", "display_name": "Opus"}, + "cost": {"total_cost_usd": 0.05, "total_duration_ms": 60000}, "context_window": { "context_window_size": 200000, "used_percentage": 45, @@ -242,6 +436,30 @@ def test_pipe_stdin(self): assert result.returncode == 0 assert "\u25d5\u203f\u25d5" in result.stdout # ◕‿◕ assert "Ctx:45%" in result.stdout + assert "$0.05" in result.stdout # exact cost + + def test_pipe_stdin_estimated_cost(self): + """Run script with no cost object — should show ~$ estimated.""" + script = os.path.join(_hooks_dir, "codingbuddy-hud.py") + stdin_data = json.dumps({ + "model": {"id": "claude-sonnet-4-5"}, + "context_window": { + "used_percentage": 10, + "current_usage": { + "input_tokens": 5000, + "cache_creation_input_tokens": 0, + "cache_read_input_tokens": 0, + }, + }, + }) + result = subprocess.run( + [sys.executable, script], + input=stdin_data, + capture_output=True, + text=True, + ) + assert result.returncode == 0 + assert "~$" in result.stdout def test_empty_stdin_fallback(self): """Script should output fallback on empty stdin."""