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 <saforem2@gmail.com>
This commit is contained in:
Brooklyn Nicholson
2026-08-28 23:27:48 -05:00
committed by brooklyn!
parent d889758521
commit fac3c62334
2 changed files with 260 additions and 28 deletions
+131 -28
View File
@@ -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 ──────────────
+129
View File
@@ -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")