refactor(hermes_cli): collapse try/except-pass to contextlib.suppress, threshold-style ladder, shared status-bar tail assembly

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:

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 <profile>` in

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-<name>`` 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"):