diff --git a/hermes_cli/checkpoints.py b/hermes_cli/checkpoints.py index f0545a5673..2d1f29c372 100644 --- a/hermes_cli/checkpoints.py +++ b/hermes_cli/checkpoints.py @@ -66,8 +66,6 @@ def _print_archives(archives) -> None: def cmd_prune(args: argparse.Namespace) -> int: from tools.checkpoint_manager import prune_checkpoints, store_status - retention_days = args.retention_days - max_size_mb = args.max_size_mb delete_orphans = not args.keep_orphans # Restricts orphan deletion to exactly the identities shown in the confirmation preview @@ -102,15 +100,15 @@ def cmd_prune(args: argparse.Namespace) -> int: orphan_allowlist.update(p["path"] for p in pre_v2_orphans) print("Pruning checkpoint store…") - print(f" retention_days: {retention_days}") + print(f" retention_days: {args.retention_days}") print(f" delete_orphans: {delete_orphans}") - print(f" max_total_size_mb: {max_size_mb}") + print(f" max_total_size_mb: {args.max_size_mb}") print() result = prune_checkpoints( - retention_days=retention_days, + retention_days=args.retention_days, delete_orphans=delete_orphans, - max_total_size_mb=max_size_mb, + max_total_size_mb=args.max_size_mb, orphan_allowlist=orphan_allowlist, ) print(f"Scanned: {result['scanned']}") @@ -123,11 +121,10 @@ def cmd_prune(args: argparse.Namespace) -> int: def _confirm(prompt: str) -> bool: try: - resp = input(f"{prompt} [y/N]: ").strip().lower() + return input(f"{prompt} [y/N]: ").strip().lower() in {"y", "yes"} except (EOFError, KeyboardInterrupt): print() return False - return resp in {"y", "yes"} def _confirmed(args: argparse.Namespace, prompt: str) -> bool: diff --git a/hermes_cli/cli_session_mixin.py b/hermes_cli/cli_session_mixin.py index bc070c0e7d..b67137ce63 100644 --- a/hermes_cli/cli_session_mixin.py +++ b/hermes_cli/cli_session_mixin.py @@ -7,6 +7,7 @@ inside each method (``from cli import ...``) — never at module load time (impo from __future__ import annotations +import contextlib import os import shutil import sys @@ -43,6 +44,12 @@ def _timestamp_or(value, default): return default +def _squash(text: str, limit: int = 120) -> str: + """Collapse whitespace and cap to *limit* cells with an ellipsis.""" + text = " ".join(text.split()) + return text[: limit - 1] + "…" if len(text) > limit else text + + def _dim_notice(cli, msg: str, quiet: bool) -> None: """Print a dim notice — raw to stderr on quiet (pre-TUI) paths. Module-level so tests can call the resume helpers unbound against a minimal stand-in.""" @@ -151,20 +158,15 @@ class CLISessionMixin: if current and os.path.realpath(recorded) == os.path.realpath(current): return if not os.path.isdir(recorded): - _dim_notice(self, - f"⚠ Session's working directory is gone: {recorded} — staying in {current or '.'}", - quiet, - ) - return - try: - os.chdir(recorded) - except OSError as e: - _dim_notice( - self, f"⚠ Could not enter session's working directory {recorded}: {e}", quiet - ) - return - os.environ["TERMINAL_CWD"] = recorded - _dim_notice(self, f"↻ Working directory: {recorded}", quiet) + msg = f"⚠ Session's working directory is gone: {recorded} — staying in {current or '.'}" + else: + try: + os.chdir(recorded) + os.environ["TERMINAL_CWD"] = recorded + msg = f"↻ Working directory: {recorded}" + except OSError as e: + msg = f"⚠ Could not enter session's working directory {recorded}: {e}" + _dim_notice(self, msg, quiet) def _restore_session_yolo(self, session_meta: dict, *, quiet: bool = False) -> None: """Re-enable YOLO bypass on resume when the session row's ``model_config.yolo_mode`` @@ -252,10 +254,8 @@ class CLISessionMixin: from cli import display_hermes_home session_meta = {} if self._session_db: - try: + with contextlib.suppress(Exception): session_meta = self._session_db.get_session(self.session_id) or {} - except Exception: - session_meta = {} title = (session_meta.get("title") or "").strip() created_at = _timestamp_or(session_meta.get("started_at"), self.session_start) @@ -273,30 +273,24 @@ class CLISessionMixin: is_running = bool(getattr(self, "_agent_running", False)) reasoning_label = None - try: - rc = getattr(agent, "reasoning_config", None) or getattr(self, "reasoning_config", None) - if isinstance(rc, dict): - if rc.get("enabled") is False: - reasoning_label = "off" - elif rc.get("effort"): - reasoning_label = str(rc.get("effort")) - show_r = getattr(self, "show_reasoning", None) - if reasoning_label and show_r is not None: - reasoning_label += f" (display: {'on' if show_r else 'off'})" - except Exception: - reasoning_label = None + rc = getattr(agent, "reasoning_config", None) or getattr(self, "reasoning_config", None) + if isinstance(rc, dict): + if rc.get("enabled") is False: + reasoning_label = "off" + elif rc.get("effort"): + reasoning_label = str(rc.get("effort")) + show_r = getattr(self, "show_reasoning", None) + if reasoning_label and show_r is not None: + reasoning_label += f" (display: {'on' if show_r else 'off'})" approval_label = None try: from tools.approval import _get_approval_mode, is_approval_bypass_active_for_session approval_label = _get_approval_mode() - try: - if is_approval_bypass_active_for_session(getattr(self, "session_key", "") or ""): - approval_label += " (YOLO bypass active)" - except Exception: - pass + if is_approval_bypass_active_for_session(getattr(self, "session_key", "") or ""): + approval_label += " (YOLO bypass active)" except Exception: - approval_label = None + pass # Context window usage: reuse the status-bar snapshot (tokens / max / percent). ctx_label = None @@ -396,8 +390,8 @@ class CLISessionMixin: if not ts: return "" try: - from datetime import datetime - return f" [{datetime.fromtimestamp(float(ts)).strftime(getattr(self, 'timestamp_format', '%H:%M'))}]" + stamp = _timestamp_or(ts, None) + return f" [{stamp.strftime(getattr(self, 'timestamp_format', '%H:%M'))}]" if stamp else "" except (ValueError, OSError, TypeError): return "" @@ -410,10 +404,9 @@ class CLISessionMixin: _cli_visible_print(f" ({hidden_tool_messages} tool {noun} hidden)") hidden_tool_messages = 0 - _cli_visible_print() - _cli_visible_print("+" + "-" * 50 + "+") - _cli_visible_print("|" + " " * 12 + "(^_^) Conversation History" + " " * 11 + "|") - _cli_visible_print("+" + "-" * 50 + "+") + rule = "+" + "-" * 50 + "+" + for line in ("", rule, "|" + " " * 12 + "(^_^) Conversation History" + " " * 11 + "|", rule): + _cli_visible_print(line) for msg in self.conversation_history: role = msg.get("role", "unknown") @@ -435,14 +428,12 @@ class CLISessionMixin: continue _cli_visible_print(f"\n [Hermes #{visible_index}]{_ts_suffix(msg)}") - tool_calls = msg.get("tool_calls") or [] + n_calls = len(msg.get("tool_calls") or []) if not content_text: suffix = "" - if tool_calls: - noun = "call" if len(tool_calls) == 1 else "calls" - preview = f"(requested {len(tool_calls)} tool {noun})" - else: - preview = "(no text response)" + preview = "(no text response)" + if n_calls: + preview = f"(requested {n_calls} tool {'call' if n_calls == 1 else 'calls'})" _cli_visible_print(f" {preview}{suffix}") flush_tool_summary() @@ -451,7 +442,7 @@ class CLISessionMixin: def _notify_session_boundary(self, event_type: str) -> None: """Fire a session-boundary plugin hook (on_session_finalize / on_session_reset). Non-blocking; errors swallowed. Safe from shutdown, /new, /reset.""" - try: + with contextlib.suppress(Exception): from hermes_cli.lifecycle import finalize_session, invoke_hook context = { @@ -463,8 +454,6 @@ class CLISessionMixin: finalize_session(**context) else: invoke_hook(event_type, **context) - except Exception: - pass def _discard_session_if_empty(self, session_id: Optional[str]) -> bool: """Drop a just-ended session row that never gained content (quit-immediately, /new, @@ -536,16 +525,12 @@ class CLISessionMixin: # /new can arrive mid-turn before _flush_messages_to_session_db() ran — flush # the current turn to the OLD session before rotating or it is silently lost. if self.agent: - try: + with contextlib.suppress(Exception): self.agent._flush_messages_to_session_db( self.conversation_history, conversation_history=self.conversation_history, ) - except Exception: - pass - try: + with contextlib.suppress(Exception): self._session_db.end_session(old_session_id, "new_session") - except Exception: - pass self._discard_session_if_empty(old_session_id) self.session_start = datetime.now() @@ -574,16 +559,14 @@ class CLISessionMixin: if hasattr(self.agent, "_last_flushed_db_idx"): self.agent._last_flushed_db_idx = 0 if hasattr(self.agent, "_todo_store"): - try: + with contextlib.suppress(Exception): from tools.todo_tool import TodoStore self.agent._todo_store = TodoStore() - except Exception: - pass if hasattr(self.agent, "_invalidate_system_prompt"): self.agent._invalidate_system_prompt() if self._session_db: - try: + with contextlib.suppress(Exception): self.agent._session_db_created = False self._session_db.create_session( session_id=self.session_id, @@ -594,29 +577,24 @@ class CLISessionMixin: }, ) self.agent._session_db_created = True - except Exception: - pass if title: title = _apply_new_session_title(self, title) # Tell memory providers the session_id rotated (reset=True flushes per-session # state) BEFORE the plugin on_session_reset hook. With old history, end-of-session # extraction and this switch are queued as ONE task on the serialized worker — # end strictly before switch, without blocking /new. No history → switch inline. - try: - _mm = getattr(self.agent, "_memory_manager", None) - if _mm is not None: - if _boundary_snapshot: - _mm.commit_session_boundary_async( - _boundary_snapshot, new_session_id=self.session_id, - parent_session_id=old_session_id or "", reason="new_session", - ) - else: - _mm.on_session_switch( - self.session_id, parent_session_id=old_session_id or "", - reset=True, reason="new_session", - ) - except Exception: - pass + _mm = getattr(self.agent, "_memory_manager", None) + with contextlib.suppress(Exception): + if _mm is not None and _boundary_snapshot: + _mm.commit_session_boundary_async( + _boundary_snapshot, new_session_id=self.session_id, + parent_session_id=old_session_id or "", reason="new_session", + ) + elif _mm is not None: + _mm.on_session_switch( + self.session_id, parent_session_id=old_session_id or "", + reset=True, reason="new_session", + ) self._notify_session_boundary("on_session_reset") if not silent: @@ -640,12 +618,11 @@ class CLISessionMixin: self._pending_resume_sessions = None if not isinstance(text, str): return False - stripped = text.strip() # Only a pure number selects; "/resume 3", titles etc. fall through. - if not stripped.isdigit(): + if not text.strip().isdigit(): return False - index = int(stripped) - if index < 1 or index > len(pending): + index = int(text.strip()) + if not 1 <= index <= len(pending): _cprint(f" Resume index {index} is out of range.") _cprint(" Use /resume with no arguments to see available sessions.") return True @@ -822,15 +799,11 @@ class CLISessionMixin: if agent is None: return if invalidate_prompt and hasattr(agent, "_invalidate_system_prompt"): - try: + with contextlib.suppress(Exception): agent._invalidate_system_prompt() - except Exception: - pass if hasattr(agent, "_last_flushed_db_idx"): - try: + with contextlib.suppress(Exception): agent._last_flushed_db_idx = len(self.conversation_history) - except Exception: - pass if hasattr(agent, "_session_messages"): agent._session_messages = self.conversation_history if hasattr(agent, "_db_flush_scan_prefix"): @@ -937,16 +910,11 @@ class CLISessionMixin: # Publish only after the durable rewind succeeds (or no store exists). self._publish_truncated_history(truncated, invalidate_prompt=True) - if self.agent is not None: - # Same hook /branch fires; rewound=True invalidates per-turn document caches. - try: - _mm = getattr(self.agent, "_memory_manager", None) - if _mm is not None and self.session_id: - _mm.on_session_switch( - self.session_id, parent_session_id="", reset=False, rewound=True - ) - except Exception: - pass + # Same hook /branch fires; rewound=True invalidates per-turn document caches. + _mm = getattr(self.agent, "_memory_manager", None) + if _mm is not None and self.session_id: + with contextlib.suppress(Exception): + _mm.on_session_switch(self.session_id, parent_session_id="", reset=False, rewound=True) turn_word = "turn" if turns_undone == 1 else "turns" print( @@ -974,12 +942,10 @@ class CLISessionMixin: """Record this terminal's live session for bare ``hermes -c``. Called whenever ``self.session_id`` is (re)assigned so a later bare ``-c`` in THIS terminal resumes this conversation's live tip. Best-effort; no-op without a terminal identity.""" - try: + with contextlib.suppress(Exception): from hermes_cli.terminal_breadcrumbs import write_breadcrumb write_breadcrumb(self.session_id) - except Exception: - pass def _transfer_session_yolo(self, old_session_id: str, new_session_id: str) -> None: """Move YOLO bypass state to a new session key when ``self.session_id`` is reassigned @@ -1068,10 +1034,8 @@ class CLISessionMixin: db = getattr(self, "_session_db", None) if db is None or not session_key or session_key == "default": return - try: + with contextlib.suppress(Exception): db.set_session_yolo(session_key, enabled) - except Exception: - pass def _manual_compress(self, cmd_original: str = ""): """Manually trigger context compression. @@ -1083,7 +1047,7 @@ class CLISessionMixin: No ``compression_enabled`` gate: that flag disables *automatic* compaction only, and the context-overflow error path directs users here when it is off. """ - if not self.conversation_history or len(self.conversation_history) < 4: + if len(self.conversation_history or ()) < 4: print("(._.) Not enough conversation to compress (need at least 4 messages).") return if not self.agent: @@ -1097,11 +1061,8 @@ class CLISessionMixin: from agent.conversation_compression import finalize_context_engine_compression_notification from agent.model_metadata import estimate_request_tokens_rough - raw_args = "" - if cmd_original: - _parts = cmd_original.strip().split(None, 1) - if len(_parts) > 1: - raw_args = _parts[1].strip() + _parts = (cmd_original or "").strip().split(None, 1) + raw_args = _parts[1].strip() if len(_parts) > 1 else "" # Strip --preview/--dry-run/--aggressive before positional parsing. raw_args, preview, aggressive = extract_compress_flags(raw_args) partial, keep_last, focus_topic = parse_partial_compress_args(raw_args) @@ -1221,9 +1182,7 @@ class CLISessionMixin: from cli import CLI_CONFIG, _DIM, _RST, _cprint if not CLI_CONFIG.get("display", {}).get("persist_prompts", True): return - detail, outcome = (" ".join(s.split()) for s in (detail, outcome)) - detail = detail[:119] + "…" if len(detail) > 120 else detail - outcome = outcome[:119] + "…" if len(outcome) > 120 else outcome + detail, outcome = (_squash(s) for s in (detail, outcome)) _cprint(f"\n{_DIM}{icon} {label}: {detail} → {outcome}{_RST}") def _clear_terminal_on_exit(self): @@ -1239,13 +1198,11 @@ class CLISessionMixin: try: stream.write("\033[3J\033[2J\033[H") stream.flush() - return except Exception: - pass - try: - os.system("cls" if os.name == "nt" else "clear") - except Exception: - pass + try: + os.system("cls" if os.name == "nt" else "clear") + except Exception: + pass def _persist_active_session_before_close(self): """Best-effort flush of the agent's live ``_session_messages`` before ``end_session()`` @@ -1343,19 +1300,16 @@ class CLISessionMixin: elapsed = datetime.now() - self.session_start hours, remainder = divmod(int(elapsed.total_seconds()), 3600) minutes, seconds = divmod(remainder, 60) + duration_str = f"{seconds}s" if hours > 0: - duration_str = f"{hours}h {minutes}m {seconds}s" + duration_str = f"{hours}h {minutes}m {duration_str}" elif minutes > 0: - duration_str = f"{minutes}m {seconds}s" - else: - duration_str = f"{seconds}s" + duration_str = f"{minutes}m {duration_str}" session_title = None if self._session_db: - try: + with contextlib.suppress(Exception): session_title = self._session_db.get_session_title(self.session_id) - except Exception: - pass print("Resume this session with:") # Session IDs are profile-constrained: non-default profiles need `-p ` in diff --git a/hermes_cli/cli_status_bar_mixin.py b/hermes_cli/cli_status_bar_mixin.py index 0c7e6fd709..ec7aeb8b7d 100644 --- a/hermes_cli/cli_status_bar_mixin.py +++ b/hermes_cli/cli_status_bar_mixin.py @@ -19,6 +19,19 @@ from typing import Any, Dict, Optional _SB = "class:status-bar" _DIM = "class:status-bar-dim" _STRONG = "class:status-bar-strong" +_AGENT_COUNTERS = ( + "session_input_tokens", "session_output_tokens", "session_cache_read_tokens", + "session_cache_write_tokens", "session_prompt_tokens", "session_completion_tokens", + "session_total_tokens", "session_api_calls", +) + + +def _threshold_style(value, ladder, fallback: str) -> str: + """First ``class:status-bar-`` whose ``value >= bound`` in a descending ladder.""" + for bound, name in ladder: + if value >= bound: + return f"class:status-bar-{name}" + return f"class:status-bar-{fallback}" def _finite(v): @@ -37,9 +50,7 @@ class CLIStatusBarMixin: return "class:status-bar-critical" if percent_used > 80: return "class:status-bar-bad" - if percent_used >= 50: - return "class:status-bar-warn" - return "class:status-bar-good" + return _threshold_style(percent_used, ((50, "warn"),), "good") def _cache_hit_rate(self, snapshot: dict, precision: int = 1) -> "tuple[float, str] | None": """Return (cache_pct, label) or None without cache data. Prefers the baseline-delta pct @@ -57,11 +68,7 @@ class CLIStatusBarMixin: def _cache_hit_rate_style(self, cache_pct: float) -> str: """Higher is better (opposite of context %).""" - if cache_pct >= 70: - return "class:status-bar-good" - if cache_pct >= 40: - return "class:status-bar-warn" - return "class:status-bar-bad" + return _threshold_style(cache_pct, ((70, "good"), (40, "warn")), "bad") @staticmethod def _battery_status_style(category: str) -> str: @@ -85,18 +92,17 @@ class CLIStatusBarMixin: except Exception: reading = None + def _detail(no_battery: str) -> str: + if reading is None: + return "" + return f" — {format_battery(reading)}" if reading.available else f" — {no_battery}" + if arg in ("status", "show"): state = "on" if self._battery_visible else "off" + detail = _detail("no battery detected on this machine") if reading is not None and reading.available: - self._console_print( - f" Battery indicator {state} — currently {format_battery(reading)}" - ) - elif reading is not None: - self._console_print( - f" Battery indicator {state} — no battery detected on this machine" - ) - else: - self._console_print(f" Battery indicator {state}") + detail = f" — currently {format_battery(reading)}" + self._console_print(f" Battery indicator {state}{detail}") return if arg in ("on", "true", "yes"): @@ -111,24 +117,16 @@ class CLIStatusBarMixin: self._battery_visible = target save_config_value("display.battery", target) - if not target: - self._console_print(" Battery indicator off") - elif reading is not None and not reading.available: + if target: self._console_print( - " Battery indicator on — no battery detected, so nothing will show here" + f" Battery indicator on{_detail('no battery detected, so nothing will show here')}" ) - elif reading is not None: - self._console_print(f" Battery indicator on — {format_battery(reading)}") else: - self._console_print(" Battery indicator on") + self._console_print(" Battery indicator off") @staticmethod def _compression_count_style(count: int) -> str: - if count >= 10: - return "class:status-bar-bad" - if count >= 5: - return "class:status-bar-warn" - return _DIM + return _threshold_style(count, ((10, "bad"), (5, "warn")), "dim") def _build_context_bar(self, percent_used: Optional[int], width: int = 10) -> str: safe_percent = max(0, min(100, percent_used or 0)) @@ -206,14 +204,7 @@ class CLIStatusBarMixin: "context_tokens": 0, "context_length": None, "context_percent": None, - "session_input_tokens": 0, - "session_output_tokens": 0, - "session_cache_read_tokens": 0, - "session_cache_write_tokens": 0, - "session_prompt_tokens": 0, - "session_completion_tokens": 0, - "session_total_tokens": 0, - "session_api_calls": 0, + **dict.fromkeys(_AGENT_COUNTERS, 0), "compressions": 0, "active_background_tasks": 0, "active_background_processes": 0, @@ -280,11 +271,7 @@ class CLIStatusBarMixin: if not agent: return snapshot - for key in ( - "session_input_tokens", "session_output_tokens", "session_cache_read_tokens", - "session_cache_write_tokens", "session_prompt_tokens", "session_completion_tokens", - "session_total_tokens", "session_api_calls", - ): + for key in _AGENT_COUNTERS: snapshot[key] = getattr(agent, key, 0) or 0 compressor = getattr(agent, "context_compressor", None) @@ -382,11 +369,12 @@ class CLIStatusBarMixin: pending = str(getattr(self, "_pending_title", None) or "").strip() session_id = str(getattr(self, "session_id", "") or "") now = time.monotonic() - if not pending: - cached_session_id = getattr(self, "_status_bar_title_session_id", None) - checked_at = float(getattr(self, "_status_bar_title_checked_at", 0.0) or 0.0) - if cached_session_id == session_id and now - checked_at < 1.5: - return str(getattr(self, "_status_bar_title_cache", "") or "") + checked_at = float(getattr(self, "_status_bar_title_checked_at", 0.0) or 0.0) + cache_fresh = ( + getattr(self, "_status_bar_title_session_id", None) == session_id and now - checked_at < 1.5 + ) + if not pending and cache_fresh: + return str(getattr(self, "_status_bar_title_cache", "") or "") title = pending db = getattr(self, "_session_db", None) if not pending and db is not None and session_id: @@ -417,19 +405,16 @@ class CLIStatusBarMixin: cw = cls._status_bar_display_width if cw(text) <= max_width: return text - ellipsis = "..." - ellipsis_width = cw(ellipsis) + ellipsis_width = cw("...") if max_width <= ellipsis_width: - return ellipsis[:max_width] - out = [] - width = 0 + return "..."[:max_width] + out, width = [], 0 for ch in text: - ch_width = cw(ch) - if width + ch_width + ellipsis_width > max_width: + if width + cw(ch) + ellipsis_width > max_width: break out.append(ch) - width += ch_width - return "".join(out).rstrip() + ellipsis + width += cw(ch) + return "".join(out).rstrip() + "..." @classmethod def _status_title_badge(cls, title: str, width: int) -> "tuple[str, int] | None": @@ -524,8 +509,7 @@ class CLIStatusBarMixin: return 0 width = width or self._get_tui_terminal_width() if width and width > 10: - import math - return max(1, math.ceil(self._status_bar_display_width(spinner_line) / width)) + return max(1, -(-self._status_bar_display_width(spinner_line) // width)) return 1 def _render_spinner_text(self) -> str: @@ -898,13 +882,8 @@ class CLIStatusBarMixin: if self._pet_anim_running: return self._pet_resolve_config() - with self._pet_lock: - kitty = ( - self._pet_enabled - and self._pet_renderer is not None - and self._pet_renderer.mode == "kitty" - ) - if kitty: + view = self._pet_view() + if view is not None and view[1]: self._pet_queue_kitty_frame() self._pet_anim_running = True self._pet_anim_thread = threading.Thread(target=self._pet_anim_loop, daemon=True) @@ -1090,19 +1069,15 @@ class CLIStatusBarMixin: segs = self._status_bar_segments( snapshot, width, field_set, self._is_session_yolo_active(), styled=False ) - parts = ["".join(t for _, t in seg) for seg in segs] - if width < 52: - battery_prefix = f"{battery_label} │ " if battery_label else "" - text = f"{battery_prefix}⚕ {model_short}" - if parts: - text = battery_prefix + " · ".join(parts) - return self._right_align_status_title(text, session_title, width) + parts = ["".join(t for _, t in seg) for seg in segs] or [f"⚕ {model_short}"] + # Narrow bars always join the battery with │; wider tiers use the tier separator. if battery_label: parts.insert(0, battery_label) - if not parts: - parts = [f"⚕ {model_short}"] - sep = " · " if width < 76 else " │ " - return self._right_align_status_title(sep.join(parts), session_title, width) + if width < 52: + text = f"{parts[0]} │ " + " · ".join(parts[1:]) if battery_label else " · ".join(parts) + else: + text = (" · " if width < 76 else " │ ").join(parts) + return self._right_align_status_title(text, session_title, width) except Exception: return f"⚕ {self.model if getattr(self, 'model', None) else 'Hermes'}" @@ -1129,14 +1104,10 @@ class CLIStatusBarMixin: ) sep = " · " if width < 76 else " │ " frags: list = [] - for seg in segs: + for seg in segs or [[(_SB, " ⚕ "), (_STRONG, snapshot["model_short"])]]: if frags: frags.append((_DIM, sep)) frags.extend(seg) - if not frags: - frags = [(_SB, " ⚕ "), (_STRONG, snapshot["model_short"])] - frags.append((_SB, " ")) - # Stash indicator (📌 N) after every width tier so a parked draft is never # invisible; before the battery prepend, and the first thing the trim drops. try: @@ -1144,12 +1115,8 @@ class CLIStatusBarMixin: except Exception: stash_indicator = "" if stash_indicator and _ok("stash"): - pieces = [(_DIM, " · "), (_STRONG, stash_indicator)] - if frags and frags[-1] == (_SB, " "): - frags[-1:-1] = pieces # keep the one-cell right margin - else: - frags.extend(pieces) - + frags.extend([(_DIM, " · "), (_STRONG, stash_indicator)]) + frags.append((_SB, " ")) # one-cell right margin # Battery is the first element when enabled: prepend ahead of the ⚕ marker. battery_label = snapshot.get("battery_label") or "" if battery_label and _ok("battery"):