diff --git a/src/dippy/dippy_statusline.py b/src/dippy/dippy_statusline.py index fcde69b0..949065a7 100755 --- a/src/dippy/dippy_statusline.py +++ b/src/dippy/dippy_statusline.py @@ -5,6 +5,7 @@ import json import os +import re import shutil import subprocess import sys @@ -113,6 +114,8 @@ def error(self, event: str, **kwargs): "changes_clean": ("white", None), "changes_dirty": ("yellow", None), "context": ("white", None), + "session": ("white", None), + "session_low": ("yellow", None), "mcp_title": ("white", None), "mcp_connected": ("green", None), "mcp_disconnected": ("red", None), @@ -216,8 +219,21 @@ def get_local_mcp_servers() -> list[str]: return [] -def get_mcp_servers() -> str | None: - """Read MCP servers from local config and cached global list.""" +def _mcp_sort_key(token: str) -> tuple[bool, str]: + """Order connected servers before disconnected ('!'), each group alphabetical.""" + name = ANSI_RE.sub("", token) + disconnected = name.startswith("!") + if disconnected: + name = name[1:] + return disconnected, name.lower() + + +def get_mcp_servers() -> tuple[str, list[str]] | None: + """Read MCP servers from local config and cached global list. + + Returns the styled "MCP:" title and the list of styled server tokens, so the + caller can lay them out with width awareness. + """ local_servers = get_local_mcp_servers() conn_r, conn_g, conn_b = hex_to_rgb(MOLOKAI[STYLES["mcp_connected"][0]]) local_styled = [ @@ -254,18 +270,21 @@ def get_mcp_servers() -> str | None: log.debug("mcp_cache_refresh_spawned", age=age, ttl=MCP_CACHE_TTL) except Exception: log.error("mcp_cache_refresh_failed") - all_servers = local_styled.copy() + server_tokens = local_styled.copy() if cached: - all_servers.append(cached) - if not all_servers: + server_tokens.extend(cached.split(", ")) + server_tokens.sort(key=_mcp_sort_key) + if not server_tokens: log.debug("mcp_no_servers") return None log.debug( - "mcp_servers_result", local_count=len(local_servers), has_cached=bool(cached) + "mcp_servers_result", + local_count=len(local_servers), + server_count=len(server_tokens), ) fg_c, bg_c = STYLES["mcp_title"] title = style("MCP:", fg_c, bg_c) - return f"{title} {', '.join(all_servers)}" + return title, server_tokens def is_dippy_configured() -> bool: @@ -294,71 +313,57 @@ def is_dippy_configured() -> bool: return False -def get_context_from_transcript(transcript_path: str) -> int | None: - """Read transcript JSONL and get actual context length from most recent message.""" - if not transcript_path: - log.debug("transcript_no_path") - return None - try: - with open(transcript_path, "rb") as f: - f.seek(0, 2) - size = f.tell() - chunk_size = min(size, 64 * 1024) - f.seek(max(0, size - chunk_size)) - lines = f.read().decode("utf-8", errors="ignore").strip().split("\n") - for line in reversed(lines): - try: - entry = json.loads(line) - usage = entry.get("message", {}).get("usage") - if usage: - total = ( - usage.get("input_tokens", 0) - + usage.get("output_tokens", 0) - + usage.get("cache_read_input_tokens", 0) - + usage.get("cache_creation_input_tokens", 0) - ) - log.debug( - "transcript_tokens_found", tokens=total, path=transcript_path - ) - return total - except json.JSONDecodeError: - continue - log.debug("transcript_no_usage", path=transcript_path, lines_checked=len(lines)) - except FileNotFoundError: - log.debug("transcript_not_found", path=transcript_path) - except Exception: - log.error("transcript_read_failed", path=transcript_path) - return None - - def get_context_remaining(data: dict) -> str | None: + """Percent of context left until auto-compact (at 80% used). + + Uses Claude Code's pre-calculated context_window.used_percentage, which is + input-only (input + cache_creation + cache_read, excluding output_tokens) to + match /context. + """ try: ctx = data.get("context_window", {}) - size = ctx.get("context_window_size", 0) - if not size: - log.debug("context_no_window_size") - return None - used = get_context_from_transcript(data.get("transcript_path", "")) - if used is None: - log.debug("context_no_usage_data", size=size) - fg_c, bg_c = STYLES["context"] - return style("ctx: 80% left", fg_c, bg_c) - used_pct = used * 100 // size - until_compact = max(0, 80 - used_pct) - log.debug( - "context_calculated", - size=size, - used=used, - used_pct=used_pct, - remaining_pct=until_compact, - ) + used_pct = ctx.get("used_percentage") fg_c, bg_c = STYLES["context"] + if used_pct is None: + log.debug("context_no_usage_data") + return style("ctx: 80% left", fg_c, bg_c) + until_compact = max(0, 80 - int(used_pct)) + log.debug("context_calculated", used_pct=used_pct, remaining_pct=until_compact) return style(f"ctx: {until_compact}% left", fg_c, bg_c) except Exception: log.error("context_remaining_failed") return None +def get_session_limit_remaining(data: dict) -> str | None: + """Percent of the 5-hour session rate-limit window remaining. + + Sourced from Claude Code's pre-calculated rate_limits.five_hour. Absent for + API billing and until the first API response of the session. + """ + try: + five_hour = data.get("rate_limits", {}).get("five_hour", {}) + used_pct = five_hour.get("used_percentage") + if used_pct is None: + log.debug("session_limit_absent") + return None + remaining = max(0, round(100 - used_pct)) + key = "session_low" if remaining <= 20 else "session" + label = f"sess: {remaining}% left" + resets_at = five_hour.get("resets_at") + if resets_at: + reset_local = datetime.fromtimestamp(resets_at) + hour = reset_local.hour % 12 or 12 + meridiem = "am" if reset_local.hour < 12 else "pm" + label = f"{label} until {hour}:{reset_local.minute:02d}{meridiem}" + log.debug("session_limit", used_pct=used_pct, remaining=remaining) + fg_c, bg_c = STYLES[key] + return style(label, fg_c, bg_c) + except Exception: + log.error("session_limit_failed") + return None + + def get_git_changes(cwd: str) -> str | None: if not cwd: log.debug("git_changes_no_cwd") @@ -423,6 +428,47 @@ def get_git_branch(cwd: str) -> str | None: return None +ANSI_RE = re.compile(r"\033\[[0-9;]*m") + + +def visible_len(text: str) -> int: + """Length of text as rendered, ignoring ANSI color escapes.""" + return len(ANSI_RE.sub("", text)) + + +def lay_out_statusline( + head: str, mcp_title: str, servers: list[str], width: int | None +) -> str: + """Join the head and MCP segment, wrapping servers to new lines past `width`. + + Keeps everything on one line when it fits. Otherwise packs servers after the + inline "MCP:" title and spills the overflow onto continuation lines, dropping + the whole MCP block to its own line when even the first server won't fit. + """ + sep = " | " + one_line = f"{head}{sep}{mcp_title} {', '.join(servers)}" + if width is None or visible_len(one_line) <= width: + return one_line + lines: list[str] = [] + inline_start = f"{head}{sep}{mcp_title}" + if visible_len(f"{inline_start} {servers[0]}") <= width: + cur = inline_start + else: + lines.append(head) + cur = mcp_title + has_server = False + for srv in servers: + glue = " " if not has_server else ", " + if not has_server or visible_len(f"{cur}{glue}{srv}") <= width: + cur = f"{cur}{glue}{srv}" + has_server = True + else: + lines.append(f"{cur},") + cur = srv + lines.append(cur) + return "\n".join(lines) + + def build_statusline(data: dict) -> str: model = "?" cwd = "" @@ -453,11 +499,28 @@ def build_statusline(data: dict) -> str: ctx_remaining = get_context_remaining(data) if ctx_remaining: parts.append(ctx_remaining) + session_remaining = get_session_limit_remaining(data) + if session_remaining: + parts.append(session_remaining) + head = " | ".join(parts) mcp = get_mcp_servers() - if mcp: - parts.append(mcp) - log.debug("build_statusline_done", parts_count=len(parts)) - return " | ".join(parts) + if not mcp: + log.debug("build_statusline_done", parts_count=len(parts), has_mcp=False) + return head + mcp_title, servers = mcp + try: + width = int(os.environ["COLUMNS"]) - 2 + if width <= 0: + width = None + except (KeyError, ValueError): + width = None + log.debug( + "build_statusline_done", + parts_count=len(parts), + width=width, + server_count=len(servers), + ) + return lay_out_statusline(head, mcp_title, servers, width) def main():