From fac3c623345b03e20db3f3ce7bb4b10a9570281c Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Fri, 28 Aug 2026 23:27:48 -0500 Subject: [PATCH] feat(cli): render Ghostty-level pets in the interactive pane Reuse kitty Unicode placeholders plus after_render write_raw so prompt_toolkit's screen-diff can host the same crisp sprite as the TUI. Re-queue the transmit after Ctrl+L / resize so the image comes back. Co-authored-by: Sam Foreman --- cli.py | 159 +++++++++++++++++++++++++++------ tests/cli/test_cli_pet_pane.py | 129 ++++++++++++++++++++++++++ 2 files changed, 260 insertions(+), 28 deletions(-) diff --git a/cli.py b/cli.py index 7d5aec9f5b..04e7e3dbc6 100644 --- a/cli.py +++ b/cli.py @@ -55,12 +55,14 @@ from hermes_cli.cli_agent_setup_mixin import CLIAgentSetupMixin from hermes_cli.cli_commands_mixin import CLICommandsMixin from hermes_cli.cli_billing_mixin import CLIBillingMixin from agent.interrupt_compat import request_hard_interrupt +from agent.pet import render as pet_render # prompt_toolkit for fixed input area TUI from prompt_toolkit.history import FileHistory from prompt_toolkit.styles import Style as PTStyle from prompt_toolkit.patch_stdout import patch_stdout from prompt_toolkit.application import Application +from prompt_toolkit.output import ColorDepth from prompt_toolkit.layout import Layout, HSplit, Window, FormattedTextControl, ConditionalContainer, WindowAlign from prompt_toolkit.layout.processors import Processor, Transformation, PasswordProcessor, ConditionalProcessor from prompt_toolkit.filters import Condition @@ -5594,15 +5596,18 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): self._command_running = False self._command_blocks_input = False self._command_status = "" - # Petdex mascot (opt-in via display.pet). The base CLI mirrors the TUI's - # PetPane: a half-block sprite above the prompt that reacts to agent - # activity. Lazily resolved; an invalidate timer drives the animation. + # Petdex mascot (opt-in via display.pet). Kitty/Ghostty use Unicode + # placeholders plus out-of-band image transmission; other terminals + # use the truecolor half-block fallback. self._pet_renderer = None # agent.pet.render.PetRenderer | None self._pet_slug: str = "" self._pet_enabled: bool = False self._pet_cols: int = 18 self._pet_scale: float = 0.7 self._pet_frames_cache: dict = {} # state -> list[grid] + self._pet_kitty_cache: dict = {} # state -> kitty placeholder payload + self._pet_kitty_image_id: int = 0 + self._pet_kitty_pending: str = "" self._pet_frame_idx: int = 0 self._pet_lock = threading.Lock() self._pet_cfg_checked: float = 0.0 @@ -5813,6 +5818,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): if getattr(self, "_terminal_io_broken", False): return _replay_output_history() + self._pet_queue_kitty_frame() try: app.invalidate() except OSError as exc: @@ -6031,6 +6037,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): pass if new_width is not None: self._last_resize_width = new_width + if width_changed: + self._pet_queue_kitty_frame() original_on_resize() self._schedule_status_bar_unsuppress(app) @@ -6777,16 +6785,24 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): # ── Petdex mascot (base-CLI pet pane) ─────────────────────────────── # - # Parity with the TUI: a half-block sprite rendered as a prompt_toolkit - # window above the prompt, reacting to agent state and animated by a timer - # that calls ``app.invalidate()``. Half-blocks only — the crisp Kitty image - # protocol can't coexist with prompt_toolkit's patch_stdout output layer - # (raw image escapes get swallowed/mangled), so we use truecolor styled - # text, which prompt_toolkit renders natively in any 24-bit terminal. + # Parity with the TUI: a sprite in a prompt_toolkit window above the + # prompt. Kitty/Ghostty use Unicode placeholders — prompt_toolkit owns + # the measurable grid; image bytes go out-of-band as a virtual placement + # via after_render + write_raw (cursor untouched). WezTerm/iTerm/sixel + # stay on half-blocks: they are not placeholder-capable. _PET_FRAME_INTERVAL = 0.16 _PET_CFG_INTERVAL = 2.5 + def _pet_clear_runtime(self) -> None: + """Drop renderer + queued Kitty state. Caller holds ``_pet_lock``.""" + self._pet_enabled = False + self._pet_renderer = None + self._pet_frames_cache.clear() + self._pet_kitty_cache.clear() + self._pet_kitty_pending = "" + self._pet_kitty_image_id = 0 + def _pet_resolve_config(self) -> None: """(Re)resolve the active pet from config — picks up live enable/disable/ @@ -6795,7 +6811,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): """ try: from agent.pet import constants, store - from agent.pet.render import PetRenderer from hermes_cli.config import load_config cfg = load_config() @@ -6808,43 +6823,48 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): slug = str(pet_cfg.get("slug", "") or "") scale = float(pet_cfg.get("scale", constants.DEFAULT_SCALE) or constants.DEFAULT_SCALE) cols = constants.resolve_cols(scale, pet_cfg.get("unicode_cols", 0)) + configured_mode = str(pet_cfg.get("render_mode", "auto") or "auto").lower() + # Placeholders only on kitty/Ghostty. WezTerm speaks kitty APC but + # not U+10EEEE — detect_terminal_graphics() still returns kitty + # there, which is why this gate is narrower. + use_kitty = configured_mode in ("", "auto", "kitty") and pet_render.supports_kitty_placeholders() + renderer_mode = "kitty" if use_kitty else "unicode" - if not enabled: + if not enabled or configured_mode == "off": with self._pet_lock: - self._pet_enabled = False - self._pet_renderer = None - self._pet_frames_cache.clear() + self._pet_clear_runtime() return pet = store.resolve_active_pet(slug) if pet is None or not pet.exists: with self._pet_lock: - self._pet_enabled = False - self._pet_renderer = None - self._pet_frames_cache.clear() + self._pet_clear_runtime() return with self._pet_lock: - # Rebuild only when the resolved pet or geometry changes. + # Rebuild only when the resolved pet, mode, or geometry changes. if ( self._pet_renderer is None or self._pet_slug != pet.slug or self._pet_cols != cols or self._pet_scale != scale + or self._pet_renderer.mode != renderer_mode ): - self._pet_renderer = PetRenderer( - str(pet.spritesheet), mode="unicode", scale=scale, unicode_cols=cols + self._pet_renderer = pet_render.PetRenderer( + str(pet.spritesheet), mode=renderer_mode, scale=scale, unicode_cols=cols ) self._pet_slug = pet.slug self._pet_cols = cols self._pet_scale = scale self._pet_frames_cache.clear() + self._pet_kitty_cache.clear() + self._pet_kitty_pending = "" + self._pet_kitty_image_id = pet_render.kitty_image_id(pet.slug) self._pet_frame_idx = 0 self._pet_enabled = True except Exception: with self._pet_lock: - self._pet_enabled = False - self._pet_renderer = None + self._pet_clear_runtime() def _pet_flash(self, state: str, secs: float = 1.6) -> None: """Briefly force a transient reaction (wave/jump/failed) before resting.""" @@ -6919,12 +6939,73 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): self._pet_frames_cache[state] = grids return grids + def _pet_kitty_payload_for(self, state: str) -> dict | None: + """Return and cache a Kitty virtual-placeholder payload for *state*.""" + with self._pet_lock: + cached = self._pet_kitty_cache.get(state) + if cached is not None: + return cached + renderer = self._pet_renderer + image_id = self._pet_kitty_image_id + if renderer is None or renderer.mode != "kitty": + return None + try: + # PNG encoding is outside _pet_lock: first visit of a state must + # not stall the prompt under the lock. + payload = renderer.kitty_payload(state, image_id=image_id) + except Exception: + payload = None + if payload is not None: + payload = {**payload, "image_id": image_id} + with self._pet_lock: + if self._pet_renderer is renderer and self._pet_kitty_image_id == image_id: + self._pet_kitty_cache[state] = payload + return payload + + def _pet_queue_kitty_frame(self, state: str | None = None) -> None: + """Queue one virtual Kitty frame for the next prompt_toolkit render.""" + if state is None: + state = self._derive_pet_state() + payload = self._pet_kitty_payload_for(state) + if not payload or not payload.get("frames"): + return + with self._pet_lock: + if self._pet_renderer is not None and self._pet_renderer.mode == "kitty": + self._pet_kitty_pending = payload["frames"][self._pet_frame_idx % len(payload["frames"])] + + def _pet_flush_kitty_frame(self, app) -> None: + """Write a queued APC after prompt_toolkit has finished its screen diff.""" + with self._pet_lock: + frame = self._pet_kitty_pending + self._pet_kitty_pending = "" + if not frame: + return + try: + # U=1/q=2 leaves the cursor and input stream untouched. + app.output.write_raw(frame) + app.output.flush() + except (OSError, ValueError): + pass + def _pet_fragments(self): """Return prompt_toolkit FormattedText for the current pet frame, or [].""" with self._pet_lock: if not self._pet_enabled or self._pet_renderer is None: return [] state = self._derive_pet_state() + kitty = self._pet_renderer.mode == "kitty" + if kitty: + payload = self._pet_kitty_payload_for(state) + if not payload: + return [] + color = pet_render.kitty_color_hex(payload["image_id"]) + frags = [] + for y, row in enumerate(payload["placeholder"]): + if y: + frags.append(("", "\n")) + frags.append((f"fg:{color}", row)) + return frags + with self._pet_lock: grids = self._pet_frames_for(state) if not grids: return [] @@ -6956,7 +7037,13 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): with self._pet_lock: if not self._pet_enabled or self._pet_renderer is None: return 0 - grids = self._pet_frames_for(self._derive_pet_state()) + state = self._derive_pet_state() + kitty = self._pet_renderer.mode == "kitty" + if kitty: + payload = self._pet_kitty_payload_for(state) + return int(payload.get("rows", 0)) if payload else 0 + with self._pet_lock: + grids = self._pet_frames_for(state) if not grids or not grids[0]: return 0 return len(grids[0]) @@ -6976,6 +7063,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): continue with self._pet_lock: self._pet_frame_idx += 1 + kitty = self._pet_renderer is not None and self._pet_renderer.mode == "kitty" + if kitty: + self._pet_queue_kitty_frame() app = getattr(self, "_app", None) if app is not None: try: @@ -6991,6 +7081,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): 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: + self._pet_queue_kitty_frame() self._pet_anim_running = True self._pet_anim_thread = threading.Thread(target=self._pet_anim_loop, daemon=True) self._pet_anim_thread.start() @@ -19485,10 +19579,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): wrap_lines=True, ) - # Petdex mascot — right-aligned half-block sprite above the prompt, - # mirroring the TUI's PetPane. Collapses to height 0 when no pet is - # enabled, so it's a no-op for everyone else. The _pet_anim_loop thread - # advances frames + invalidates; align=RIGHT pins it to the edge. + # Petdex mascot — right-aligned Kitty placeholder or half-block sprite + # above the prompt. Collapses to height 0 when no pet is enabled. + # The animation thread queues virtual Kitty frames; after_render + # writes them out-of-band while prompt_toolkit owns the placeholder grid. self._pet_widget = Window( content=FormattedTextControl(self._pet_fragments), height=self._pet_widget_height, @@ -20275,7 +20369,10 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): # in _strip_leaked_terminal_responses still guards residual leaks. _cpr_disabled_output = _select_classic_cli_pt_output(sys.stdout) - # Create the application + # Kitty placeholders encode their image id in exact foreground RGB, so + # placeholder-capable terminals (kitty/Ghostty) use 24-bit color for + # the whole prompt_toolkit application — quantizing only that pane + # is not supported. WezTerm is excluded: it is not placeholder-capable. app = Application( layout=layout, key_bindings=kb, @@ -20283,6 +20380,11 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): full_screen=False, mouse_support=False, **({"output": _cpr_disabled_output} if _cpr_disabled_output is not None else {}), + **( + {"color_depth": ColorDepth.DEPTH_24_BIT} + if pet_render.supports_kitty_placeholders() + else {} + ), # Read from display.cli_refresh_interval (default 0 = disabled). # When non-zero, prompt_toolkit redraws the UI on this cadence # during idle, keeping wall-clock status-bar read-outs ticking. @@ -20304,6 +20406,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin): **({'cursor': _STEADY_CURSOR} if _STEADY_CURSOR is not None else {}), ) _disable_prompt_toolkit_cpr_warning(app) + app.after_render += self._pet_flush_kitty_frame self._app = app # Store reference for clarify_callback # ── Fix ghost status-bar lines on terminal resize ────────────── diff --git a/tests/cli/test_cli_pet_pane.py b/tests/cli/test_cli_pet_pane.py index f848971d85..9f42508b0d 100644 --- a/tests/cli/test_cli_pet_pane.py +++ b/tests/cli/test_cli_pet_pane.py @@ -53,6 +53,9 @@ def _make_cli(): cli_obj._pet_cols = 18 cli_obj._pet_scale = 0.7 cli_obj._pet_frames_cache = {} + cli_obj._pet_kitty_cache = {} + cli_obj._pet_kitty_image_id = 0 + cli_obj._pet_kitty_pending = "" cli_obj._pet_frame_idx = 0 cli_obj._agent_running = False # Transient-beat + reasoning state (set by HermesCLI.__init__ in production). @@ -134,3 +137,129 @@ def test_pet_resolve_config_enables_and_disables(boba_like): cli_obj._pet_resolve_config() assert cli_obj._pet_enabled is False assert cli_obj._pet_renderer is None + + +def test_pet_fragments_render_kitty_placeholders(boba_like): + from agent.pet import render + + cli_obj = _make_cli() + pet = store.load_pet("boba") + assert pet is not None + cli_obj._pet_renderer = PetRenderer(str(pet.spritesheet), mode="kitty", scale=0.4) + cli_obj._pet_slug = "boba" + cli_obj._pet_kitty_image_id = render.kitty_image_id("boba") + cli_obj._pet_enabled = True + + frags = cli_obj._pet_fragments() + assert frags + assert any("\U0010eeee" in text for _, text in frags) + payload = cli_obj._pet_kitty_payload_for("idle") + assert payload is not None + color = render.kitty_color_hex(payload["image_id"]) + assert all(f"fg:{color}" in style for style, text in frags if text != "\n") + assert cli_obj._pet_widget_height() > 0 + + cli_obj._pet_queue_kitty_frame("idle") + assert cli_obj._pet_kitty_pending.startswith("\x1b_G") + + class Output: + def __init__(self): + self.raw = "" + self.flushed = False + + def write_raw(self, text): + self.raw += text + + def flush(self): + self.flushed = True + + class App: + output = Output() + + app = App() + cli_obj._pet_flush_kitty_frame(app) + assert app.output.raw.startswith("\x1b_G") + assert app.output.flushed is True + assert cli_obj._pet_kitty_pending == "" + + +def test_pet_off_clears_pending_kitty_frame(boba_like): + from hermes_cli.config import load_config, save_config + + cli_obj = _make_cli() + cli_obj._pet_kitty_pending = "stale-apc" + cfg = load_config() + cfg.setdefault("display", {}).setdefault("pet", {}).update( + {"enabled": True, "slug": "boba", "render_mode": "off"} + ) + save_config(cfg) + + cli_obj._pet_resolve_config() + + assert cli_obj._pet_enabled is False + assert cli_obj._pet_renderer is None + assert cli_obj._pet_kitty_pending == "" + + +def test_pet_resolve_wezterm_stays_unicode(boba_like, monkeypatch): + from hermes_cli.config import load_config, save_config + + monkeypatch.delenv("KITTY_WINDOW_ID", raising=False) + monkeypatch.setenv("TERM", "xterm-256color") + monkeypatch.setenv("TERM_PROGRAM", "WezTerm") + cli_obj = _make_cli() + cfg = load_config() + cfg.setdefault("display", {}).setdefault("pet", {}).update( + {"enabled": True, "slug": "boba", "render_mode": "auto"} + ) + save_config(cfg) + + cli_obj._pet_resolve_config() + + assert cli_obj._pet_renderer is not None + assert cli_obj._pet_renderer.mode == "unicode" + + +def test_pet_resolve_ghostty_uses_kitty(boba_like, monkeypatch): + from hermes_cli.config import load_config, save_config + + monkeypatch.delenv("WEZTERM_PANE", raising=False) + monkeypatch.delenv("KITTY_WINDOW_ID", raising=False) + monkeypatch.setenv("TERM", "xterm-ghostty") + monkeypatch.setenv("TERM_PROGRAM", "ghostty") + cli_obj = _make_cli() + cfg = load_config() + cfg.setdefault("display", {}).setdefault("pet", {}).update( + {"enabled": True, "slug": "boba", "render_mode": "auto"} + ) + save_config(cfg) + + cli_obj._pet_resolve_config() + + assert cli_obj._pet_renderer is not None + assert cli_obj._pet_renderer.mode == "kitty" + + +def test_force_full_redraw_requeues_kitty_frame(boba_like, monkeypatch): + from agent.pet import render + + cli_obj = _make_cli() + pet = store.load_pet("boba") + assert pet is not None + cli_obj._pet_renderer = PetRenderer(str(pet.spritesheet), mode="kitty", scale=0.4) + cli_obj._pet_enabled = True + cli_obj._pet_kitty_image_id = render.kitty_image_id("boba") + cli_obj._terminal_io_broken = False + cli_obj._clear_prompt_toolkit_screen = lambda *a, **k: None + cli_obj._redraw_rebuilds_scrollback = lambda: False + + class App: + def invalidate(self): + self.invalidated = True + + cli_obj._app = App() + monkeypatch.setattr("cli._replay_output_history", lambda: None) + + cli_obj._force_full_redraw() + + assert cli_obj._pet_kitty_pending.startswith("\x1b_G")