38f97640c0
Live tmux probes showed the painted-width budget from the previous commit duplicating rows on reflowing terminals: a shrink re-wraps the rows the replay counted at their old width (Rich pads lines to the full width, so every reply line grows) and pushes the extra into scrollback. Terminals that keep rows in place (xterm) need the opposite count, and the app cannot see which kind it runs in. - A shrink replays only on a reflowing terminal (_terminal_reflows: TMUX yes; STY, XTERM_VERSION, TERM=linux no; unknown counts as reflowing), counting rows at the new width and the old chrome as re-wrapped to it. Elsewhere the transcript stays as the terminal truncated it and prompt_toolkit's own erase covers the chrome: nothing is lost. - Ctrl+L counts rows at their painted width only where rows stay put. - _cprint output is held from the resize signal until its recovery ran: a paint in between erased the re-wrapped chrome from prompt_toolkit's stale cursor, leaving two rows of it in the transcript that the replay could not account for (two duplicated lines per shrink).
844 lines
32 KiB
Python
844 lines
32 KiB
Python
"""Classic-CLI rendering helpers: reasoning-tag stripping, ANSI/skin colours, light-mode detection, markdown/final-content rendering, output-history replay, ``_cprint`` and the panel box/wrap helpers.
|
|
|
|
Split out of ``cli.py``; ``cli`` re-exports every public name and moved bodies late-bind
|
|
cli-level names through ``from cli import ...`` at call time so facade monkeypatch seams hold.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import functools
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import textwrap
|
|
import threading
|
|
import time
|
|
from contextlib import contextmanager, suppress
|
|
from hermes_cli.banner import format_banner_version_label
|
|
from rich.console import Console
|
|
from rich.text import Text as _RichText
|
|
from typing import Any
|
|
|
|
def _cli():
|
|
"""Late import of the ``cli`` facade: mutable CLI module state (and its test seams) lives there."""
|
|
import cli
|
|
|
|
return cli
|
|
|
|
|
|
_REASONING_TAGS = ("REASONING_SCRATCHPAD", "think", "thinking", "reasoning", "thought")
|
|
|
|
|
|
_TOOL_CALL_TAGS = ("tool_call", "tool_calls", "tool_result", "function_call", "function_calls")
|
|
|
|
|
|
def _strip_reasoning_tags(text: str) -> str:
|
|
"""Strip reasoning blocks (closed, unterminated, orphan-close) and leaked tool-call XML from display text.
|
|
|
|
Keep in sync with ``agent.agent_runtime_helpers.strip_think_blocks`` and the stream consumer's think-tag sets.
|
|
|
|
Also strips tool-call XML blocks some open models leak into visible content (``<tool_call>``,
|
|
``<function_calls>``, Gemma-style ``<function name="…">…</function>``). Ported from
|
|
openclaw/openclaw#67318.
|
|
"""
|
|
from cli import _REASONING_TAGS, _TOOL_CALL_TAGS
|
|
cleaned = text
|
|
for tag in _REASONING_TAGS:
|
|
cleaned = re.sub(rf"<{tag}>.*?</{tag}>\s*", "", cleaned, flags=re.DOTALL | re.IGNORECASE)
|
|
cleaned = re.sub(rf"<{tag}>.*$", "", cleaned, flags=re.DOTALL | re.IGNORECASE)
|
|
cleaned = re.sub(rf"</{tag}>\s*", "", cleaned, flags=re.IGNORECASE)
|
|
for tc_tag in _TOOL_CALL_TAGS:
|
|
cleaned = re.sub(
|
|
rf"<(?:[\w.-]+:)?{tc_tag}\b[^>]*>.*?</(?:[\w.-]+:)?{tc_tag}>\s*",
|
|
"", cleaned, flags=re.DOTALL | re.IGNORECASE,
|
|
)
|
|
# <function name="..."> — boundary + attribute gated to avoid prose false positives.
|
|
cleaned = re.sub(
|
|
r'(?:(?<=^)|(?<=[\n\r.!?:]))[ \t]*<function\b[^>]*\bname\s*=[^>]*>(?:(?:(?!</function>).)*)</function>\s*',
|
|
'', cleaned, flags=re.DOTALL | re.IGNORECASE,
|
|
)
|
|
cleaned = re.sub(
|
|
r'</(?:(?:[\w.-]+:)?(?:tool_call|tool_calls|tool_result|function_call|function_calls|function))>\s*', '', cleaned,
|
|
flags=re.IGNORECASE,
|
|
)
|
|
# Unterminated opener / stray <arg_key>/<arg_value> markup = stream cut
|
|
# mid tool-call serialization (#101899); strip to end of text.
|
|
cleaned = re.sub(
|
|
r'(?:^|\n)[ \t]*<(?:[\w.-]+:)?(?:tool_call|tool_calls|tool_result|function_call|function_calls)\b[^>]*>.*$'
|
|
r'|(?:^|\n)[^\n<]*</?arg_(?:key|value)\b.*$',
|
|
'',
|
|
cleaned,
|
|
flags=re.DOTALL | re.IGNORECASE,
|
|
)
|
|
return cleaned.strip()
|
|
|
|
|
|
def _assistant_content_as_text(content: Any) -> str:
|
|
if content is None:
|
|
return ""
|
|
if isinstance(content, str):
|
|
return content
|
|
if isinstance(content, list):
|
|
parts = [str(part.get("text", "")) for part in content if isinstance(part, dict) and part.get("type") == "text"]
|
|
return "\n".join(p for p in parts if p)
|
|
return str(content)
|
|
|
|
|
|
def _assistant_copy_text(content: Any) -> str:
|
|
from cli import _assistant_content_as_text, _strip_reasoning_tags
|
|
return _strip_reasoning_tags(_assistant_content_as_text(content))
|
|
|
|
|
|
_ACCENT_ANSI_DEFAULT = "\033[1;38;2;255;215;0m" # #FFD700 bold fallback
|
|
|
|
|
|
_BOLD = "\033[1m"
|
|
|
|
|
|
_RST = "\033[0m"
|
|
|
|
|
|
_STREAM_PAD = "" # no indent: leading whitespace pollutes copy/paste
|
|
|
|
|
|
_STREAM_PARTIAL_PREVIEW_LEN = 60 # tail of an unfinished line mirrored into the spinner
|
|
|
|
|
|
def _hex_to_ansi(hex_color: str, *, bold: bool = False) -> str:
|
|
"""Convert '#RRGGBB' to a true-color ANSI escape, remapping dark-tuned colors in light mode."""
|
|
from cli import _ACCENT_ANSI_DEFAULT, _maybe_remap_for_light_mode
|
|
hex_color = _maybe_remap_for_light_mode(hex_color)
|
|
try:
|
|
r, g, b = (int(hex_color[i:i + 2], 16) for i in (1, 3, 5))
|
|
return f"\033[{'1;' if bold else ''}38;2;{r};{g};{b}m"
|
|
except (ValueError, IndexError):
|
|
return _ACCENT_ANSI_DEFAULT if bold else "\033[38;2;184;134;11m"
|
|
|
|
|
|
_TRUE_RE = re.compile(r"^(1|true|on|yes|y)$")
|
|
|
|
|
|
_FALSE_RE = re.compile(r"^(0|false|off|no|n)$")
|
|
|
|
|
|
_LIGHT_DEFAULT_TERM_PROGRAMS = frozenset() # Apple_Terminal isn't reliable; require explicit config
|
|
|
|
|
|
def _luminance_from_hex(hex_str: str) -> float | None:
|
|
"""Rec.709 luma in [0, 1] for '#RGB'/'#RRGGBB', or None when malformed."""
|
|
s = (hex_str or "").strip().lstrip("#")
|
|
if len(s) == 3:
|
|
s = "".join(c * 2 for c in s)
|
|
if len(s) != 6 or not all(c in "0123456789abcdefABCDEF" for c in s):
|
|
return None
|
|
try:
|
|
r, g, b = int(s[0:2], 16), int(s[2:4], 16), int(s[4:6], 16)
|
|
except ValueError:
|
|
return None
|
|
return (0.2126 * r + 0.7152 * g + 0.0722 * b) / 255.0
|
|
|
|
|
|
_DA1_REPLY_RE = re.compile(rb"\x1b\[\?[0-9;]*c")
|
|
|
|
|
|
def _query_osc11_background() -> str | None:
|
|
"""Terminal background via OSC 11 as "#RRGGBB", or None.
|
|
|
|
Fenced with a DA1 sentinel (``ESC[c``): terminals answer in order and virtually all
|
|
answer DA1, so its reply proves our OSC 11 was processed — otherwise a late reply
|
|
leaks into prompt_toolkit's stdin as typed text. Skipped over SSH (round-trip too
|
|
slow; a late BEL reads as Ctrl+G). A 50 ms drain after TCSAFLUSH catches stragglers.
|
|
|
|
After the main read + TCSAFLUSH, a short drain window (50 ms) catches late-arriving bytes that slipped
|
|
past the flush — a race observed on VPS and container terminals under load (#40250).
|
|
"""
|
|
from cli import _DA1_REPLY_RE
|
|
if not sys.stdin.isatty() or not sys.stdout.isatty():
|
|
return None
|
|
if any(os.environ.get(v) for v in ("SSH_CONNECTION", "SSH_CLIENT", "SSH_TTY")):
|
|
return None
|
|
try:
|
|
import select
|
|
import termios
|
|
import tty
|
|
fd = sys.stdin.fileno()
|
|
old = termios.tcgetattr(fd)
|
|
except Exception:
|
|
return None
|
|
try:
|
|
try:
|
|
tty.setcbreak(fd)
|
|
except Exception:
|
|
return None
|
|
try:
|
|
# One write so the OSC 11 query and DA1 fence cannot reorder.
|
|
sys.stdout.write("\x1b]11;?\x1b\\\x1b[c")
|
|
sys.stdout.flush()
|
|
except Exception:
|
|
return None
|
|
# Read until the DA1 fence closes; the 1s deadline only covers terminals ignoring DA1.
|
|
deadline = time.monotonic() + 1.0
|
|
buf = b""
|
|
while time.monotonic() < deadline:
|
|
r, _, _ = select.select([fd], [], [], deadline - time.monotonic())
|
|
if not r:
|
|
continue
|
|
try:
|
|
chunk = os.read(fd, 64)
|
|
except OSError:
|
|
break
|
|
if not chunk:
|
|
break
|
|
buf += chunk
|
|
if _DA1_REPLY_RE.search(buf):
|
|
break
|
|
# Reply: \x1b]11;rgb:RRRR/GGGG/BBBB\x1b\\ — components are 1-4 hex digits.
|
|
m = re.search(rb"rgb:([0-9a-fA-F]+)/([0-9a-fA-F]+)/([0-9a-fA-F]+)", buf)
|
|
if not m:
|
|
return None
|
|
|
|
def norm(h: bytes) -> int:
|
|
v = int(h, 16)
|
|
bits = len(h) * 4
|
|
return (v * 255) // ((1 << bits) - 1) if bits else 0
|
|
r, g, b = norm(m.group(1)), norm(m.group(2)), norm(m.group(3))
|
|
return f"#{r:02X}{g:02X}{b:02X}"
|
|
finally:
|
|
# TCSAFLUSH discards unread input, scrubbing a partial reply before prompt_toolkit reads it.
|
|
with suppress(Exception):
|
|
termios.tcsetattr(fd, termios.TCSAFLUSH, old)
|
|
try:
|
|
drain_deadline = time.monotonic() + 0.05
|
|
while time.monotonic() < drain_deadline:
|
|
r, _, _ = select.select([fd], [], [], drain_deadline - time.monotonic())
|
|
if not r or not os.read(fd, 64):
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _heal_cooked_mode_drift(fd: int) -> bool:
|
|
"""Re-apply raw mode on *fd* when termios drifted back to cooked (POSIX only).
|
|
|
|
A lost ``run_in_terminal`` cooked_mode() restore makes the kernel line-buffer every
|
|
keystroke and the CLI looks dead. Mirrors prompt_toolkit's raw_mode flag surgery in
|
|
place. Returns True when healed; False when already raw or not inspectable.
|
|
"""
|
|
try:
|
|
import termios
|
|
attrs = termios.tcgetattr(fd)
|
|
except Exception:
|
|
return False
|
|
lflag = attrs[3]
|
|
if not (lflag & (termios.ICANON | termios.ECHO)):
|
|
return False # still raw — nothing to do
|
|
attrs[3] = lflag & ~(termios.ECHO | termios.ICANON | termios.IEXTEN | termios.ISIG)
|
|
attrs[0] = attrs[0] & ~(termios.IXON | termios.IXOFF | termios.ICRNL | termios.INLCR | termios.IGNCR)
|
|
attrs[6][termios.VMIN] = 1
|
|
try:
|
|
termios.tcsetattr(fd, termios.TCSANOW, attrs)
|
|
except Exception:
|
|
return False
|
|
return True
|
|
|
|
|
|
def _detect_light_mode_uncached() -> bool:
|
|
"""The detection ladder documented above; may raise (caller maps errors to dark)."""
|
|
from cli import _FALSE_RE, _LIGHT_DEFAULT_TERM_PROGRAMS, _TRUE_RE, _luminance_from_hex, _query_osc11_background
|
|
for var in ("HERMES_LIGHT", "HERMES_TUI_LIGHT"):
|
|
v = (os.environ.get(var) or "").strip().lower()
|
|
if _TRUE_RE.match(v):
|
|
return True
|
|
if _FALSE_RE.match(v):
|
|
return False
|
|
theme = (os.environ.get("HERMES_TUI_THEME") or "").strip().lower()
|
|
if theme == "light":
|
|
return True
|
|
if theme == "dark":
|
|
return False
|
|
bg_lum = _luminance_from_hex(os.environ.get("HERMES_TUI_BACKGROUND") or "")
|
|
if bg_lum is not None:
|
|
return bg_lum >= 0.5
|
|
last = (os.environ.get("COLORFGBG") or "").strip().split(";")[-1]
|
|
if last.isdigit() and 0 <= int(last) < 16:
|
|
return int(last) in {7, 15}
|
|
bg_color = _query_osc11_background()
|
|
if bg_color:
|
|
lum = _luminance_from_hex(bg_color)
|
|
if lum is not None:
|
|
return lum >= 0.5
|
|
return (os.environ.get("TERM_PROGRAM") or "").strip() in _LIGHT_DEFAULT_TERM_PROGRAMS
|
|
|
|
|
|
# Light-mode equivalents of skin colors unreadable on cream backgrounds. Only colors used
|
|
# as STANDALONE foregrounds: ones paired with a dark bg (status bar text on #1a1a2e) would
|
|
# become invisible the other direction, hence #C0C0C0/#888888/#555555/#8B8682 are skipped.
|
|
_LIGHT_MODE_REMAP: dict[str, str] = {
|
|
"#FFF8DC": "#1A1A1A", "#FFD700": "#9A6B00", "#FFBF00": "#8A5A00", "#B8860B": "#5C4500",
|
|
"#DAA520": "#6B4F00", "#F1E6CF": "#1A1A1A", "#c9d1d9": "#24292F", "#EAF7FF": "#0F1B26",
|
|
"#F5F5F5": "#1A1A1A", "#FFF0D4": "#1A1A1A", "#CD7F32": "#8A4F1A", "#FFEFB5": "#3A2A00",
|
|
}
|
|
|
|
|
|
_LIGHT_MODE_REMAP_UPPER = {k.upper(): v for k, v in _LIGHT_MODE_REMAP.items()}
|
|
|
|
|
|
def _maybe_remap_for_light_mode(hex_color: str) -> str:
|
|
"""In light mode, remap a dark-tuned color to its higher-contrast equivalent."""
|
|
from cli import _LIGHT_MODE_REMAP_UPPER, _detect_light_mode
|
|
if not _detect_light_mode():
|
|
return hex_color
|
|
if not hex_color or not hex_color.startswith("#"):
|
|
return hex_color
|
|
return _LIGHT_MODE_REMAP_UPPER.get(hex_color.upper(), hex_color)
|
|
|
|
|
|
def _install_skin_light_mode_hook() -> None:
|
|
"""Wrap SkinConfig.get_color so EVERY skin color read goes through the light-mode remap. Idempotent."""
|
|
from cli import _maybe_remap_for_light_mode
|
|
try:
|
|
from hermes_cli.skin_engine import SkinConfig # type: ignore[import]
|
|
except Exception:
|
|
return
|
|
if getattr(SkinConfig, "_hermes_light_mode_hook_installed", False):
|
|
return
|
|
_orig_get_color = SkinConfig.get_color
|
|
|
|
def _wrapped_get_color(self, key, fallback=""):
|
|
value = _orig_get_color(self, key, fallback)
|
|
try:
|
|
return _maybe_remap_for_light_mode(value)
|
|
except Exception:
|
|
return value
|
|
|
|
SkinConfig.get_color = _wrapped_get_color # type: ignore[method-assign]
|
|
SkinConfig._hermes_light_mode_hook_installed = True # type: ignore[attr-defined]
|
|
|
|
|
|
class _SkinAwareAnsi:
|
|
"""Lazy ANSI escape resolved from the skin on first use; ``.reset()`` after a ``/skin`` switch."""
|
|
|
|
def __init__(self, skin_key: str, fallback_hex: str = "#FFD700", *, bold: bool = False):
|
|
self._skin_key = skin_key
|
|
self._fallback_hex = fallback_hex
|
|
self._bold = bold
|
|
self._cached: str | None = None
|
|
|
|
def __str__(self) -> str:
|
|
from cli import _hex_to_ansi
|
|
if self._cached is None:
|
|
try:
|
|
from hermes_cli.skin_engine import get_active_skin
|
|
self._cached = _hex_to_ansi(
|
|
get_active_skin().get_color(self._skin_key, self._fallback_hex),
|
|
bold=self._bold,
|
|
)
|
|
except Exception:
|
|
self._cached = _hex_to_ansi(self._fallback_hex, bold=self._bold)
|
|
return self._cached
|
|
|
|
def __add__(self, other: str) -> str:
|
|
return str(self) + other
|
|
|
|
def __radd__(self, other: str) -> str:
|
|
return other + str(self)
|
|
|
|
def reset(self) -> None:
|
|
"""Clear cache so the next access re-reads the skin."""
|
|
self._cached = None
|
|
|
|
|
|
_ACCENT = _SkinAwareAnsi("response_border", "#FFD700", bold=True)
|
|
|
|
|
|
# dim+italic attributes (not a hex) so dim text inherits the terminal foreground in both modes.
|
|
_DIM = "\x1b[2;3m"
|
|
|
|
|
|
def _tty_wrap(s: str, sgr: str) -> str:
|
|
"""Wrap *s* in an SGR attribute when stdout is a real TTY; plain text otherwise."""
|
|
try:
|
|
return f"{sgr}{s}\x1b[0m" if sys.stdout.isatty() else str(s)
|
|
except Exception:
|
|
return str(s)
|
|
|
|
|
|
_b = functools.partial(_tty_wrap, sgr="\x1b[1m") # bold when stdout is a real TTY
|
|
|
|
|
|
_d = functools.partial(_tty_wrap, sgr="\x1b[2;3m") # dim-italic when stdout is a real TTY
|
|
|
|
|
|
def _accent_hex() -> str:
|
|
"""Return the active skin accent color for legacy CLI output lines."""
|
|
try:
|
|
from hermes_cli.skin_engine import get_active_skin
|
|
return get_active_skin().get_color("ui_accent", "#FFBF00")
|
|
except Exception:
|
|
return "#FFBF00"
|
|
|
|
|
|
def _rich_text_from_ansi(text: str) -> _RichText:
|
|
"""Rich Text from ANSI output; literal ``[brackets]`` are not treated as markup."""
|
|
return _RichText.from_ansi(text or "")
|
|
|
|
|
|
def _strip_markdown_syntax(text: str) -> str:
|
|
"""Best-effort markdown marker removal for plain-text display."""
|
|
from cli import _rich_text_from_ansi
|
|
plain = _rich_text_from_ansi(text or "").plain
|
|
# HR markers: "-"/"_" runs of 3+, but "*" only when exactly 3 (cron schedules "* * * * *").
|
|
plain = re.sub(r"^\s{0,3}(?:[-_]\s*){3,}$", "", plain, flags=re.MULTILINE)
|
|
plain = re.sub(r"^\s{0,3}(?:\*\s*){3}\s*$", "", plain, flags=re.MULTILINE)
|
|
plain = re.sub(r"^\s{0,3}#{1,6}\s+", "", plain, flags=re.MULTILINE)
|
|
# Blockquotes, lists, and checkboxes are preserved because they carry structure.
|
|
plain = re.sub(r"(```+|~~~+)", "", plain)
|
|
plain = re.sub(r"`([^`]*)`", r"\1", plain)
|
|
plain = re.sub(r"!\[([^\]]*)\]\([^\)]*\)", r"\1", plain)
|
|
plain = re.sub(r"\[([^\]]+)\]\([^\)]*\)", r"\1", plain)
|
|
plain = re.sub(r"\*\*\*([^*]+)\*\*\*", r"\1", plain)
|
|
plain = re.sub(r"(?<!\w)___([^_]+)___(?!\w)", r"\1", plain)
|
|
plain = re.sub(r"\*\*([^*]+)\*\*", r"\1", plain)
|
|
plain = re.sub(r"(?<!\w)__([^_]+)__(?!\w)", r"\1", plain)
|
|
# `*emphasis*` only when the inner text is non-whitespace (cron expressions again).
|
|
plain = re.sub(r"\*([^\s*][^*]*?[^\s*])\*", r"\1", plain)
|
|
plain = re.sub(r"(?<!\w)_([^_]+)_(?!\w)", r"\1", plain)
|
|
plain = re.sub(r"~~([^~]+)~~", r"\1", plain)
|
|
plain = re.sub(r"\n{3,}", "\n\n", plain)
|
|
return plain.strip("\n")
|
|
|
|
|
|
_WINDOWS_PATH_WITH_DOT_SEGMENT_RE = re.compile(r"(?i)(?:\b[a-z]:\\|\\\\)[^\s`]*\\\.[^\s`]*")
|
|
|
|
|
|
def _preserve_windows_dot_segments_for_markdown(text: str) -> str:
|
|
r"""Double the ``\`` before hidden dirs in Windows paths: CommonMark reads ``\.`` as an escaped dot."""
|
|
from cli import _WINDOWS_PATH_WITH_DOT_SEGMENT_RE
|
|
if "\\." not in text:
|
|
return text
|
|
|
|
def _protect(match: re.Match[str]) -> str:
|
|
return re.sub(r"(?<!\\)\\(?=\.)", r"\\\\", match.group(0))
|
|
|
|
return _WINDOWS_PATH_WITH_DOT_SEGMENT_RE.sub(_protect, text)
|
|
|
|
|
|
def _terminal_columns() -> int:
|
|
try:
|
|
return shutil.get_terminal_size((80, 24)).columns
|
|
except Exception:
|
|
return 80
|
|
|
|
|
|
def _terminal_width_for_streaming() -> int:
|
|
"""Display cells inside the streamed response box (small margin for resize races)."""
|
|
from cli import _STREAM_PAD, _terminal_columns
|
|
return max(20, _terminal_columns() - len(_STREAM_PAD) - 2)
|
|
|
|
|
|
def _render_final_assistant_content(text: str, mode: str = "render"):
|
|
"""Render final assistant content as markdown, stripped text, or raw text."""
|
|
from cli import _preserve_windows_dot_segments_for_markdown, _rich_text_from_ansi, _strip_markdown_syntax, _terminal_columns, realign_markdown_tables
|
|
from rich.markdown import Markdown
|
|
|
|
# 1 border cell each side + margin so resize races don't push a borderline table into soft-wrap.
|
|
panel_width = max(20, _terminal_columns() - 4)
|
|
|
|
normalized_mode = str(mode or "render").strip().lower()
|
|
if normalized_mode == "strip":
|
|
# Strip first (inline markdown changes cell width), then re-align padding.
|
|
return _RichText(realign_markdown_tables(_strip_markdown_syntax(text), panel_width))
|
|
if normalized_mode == "raw":
|
|
return _rich_text_from_ansi(text or "")
|
|
|
|
# Normalising under-padded tables up front gives narrow-panel fallbacks consistent input.
|
|
plain = _rich_text_from_ansi(text or "").plain
|
|
plain = _preserve_windows_dot_segments_for_markdown(plain)
|
|
plain = realign_markdown_tables(plain, panel_width)
|
|
return Markdown(plain)
|
|
|
|
|
|
def _post_stream_transform_output(response: str, result: dict | None) -> str:
|
|
"""Text still to display after a streamed response transform: the suffix, or the whole response when replaced."""
|
|
if not result or not result.get("response_transformed"):
|
|
return ""
|
|
|
|
original = result.get("pre_transform_response") or ""
|
|
if original and response.startswith(original):
|
|
return response[len(original):]
|
|
|
|
return f"\n[Response transformed after streaming]\n{response}"
|
|
|
|
|
|
def _coerce_output_history_limit(value) -> int:
|
|
try:
|
|
return max(10, int(value))
|
|
except (TypeError, ValueError):
|
|
return 200
|
|
|
|
|
|
def _clear_output_history() -> None:
|
|
_cli()._OUTPUT_HISTORY.clear()
|
|
|
|
|
|
def _output_history_recording() -> bool:
|
|
return _cli()._OUTPUT_HISTORY_ENABLED and not _cli()._OUTPUT_HISTORY_REPLAYING and not _cli()._OUTPUT_HISTORY_SUPPRESSED
|
|
|
|
|
|
def _record_output_history_entry(entry) -> None:
|
|
from cli import _output_history_recording
|
|
if _output_history_recording():
|
|
_cli()._OUTPUT_HISTORY.append(entry)
|
|
|
|
|
|
class _PaintedLine(str):
|
|
"""A recorded output line tagged with the terminal ``width`` it was painted at: a terminal
|
|
that does not reflow keeps the rows it wrapped into then, whatever the width is now."""
|
|
width = None
|
|
|
|
|
|
def _painted_columns():
|
|
"""The width the terminal soft-wraps a print at right now, or ``None``."""
|
|
try:
|
|
from prompt_toolkit.application import get_app_or_none
|
|
app = get_app_or_none()
|
|
if app is not None:
|
|
return app.output.get_size().columns
|
|
except Exception:
|
|
pass
|
|
try:
|
|
return os.get_terminal_size(sys.__stdout__.fileno()).columns
|
|
except (AttributeError, OSError, ValueError):
|
|
return None
|
|
|
|
|
|
def _record_output_history(text: str, *, force: bool = False) -> None:
|
|
"""Record ``text`` as painted now. ``force`` skips the recording check when the caller
|
|
made it at print-request time (the print itself was deferred to the app loop)."""
|
|
from cli import _output_history_recording
|
|
if force or _output_history_recording():
|
|
width = _painted_columns()
|
|
lines = []
|
|
# One entry per printed line: ``_pt_print`` ends every text with a newline, so "" and a
|
|
# trailing "\n" are blank rows on screen and must count in the replay's row budget.
|
|
for line in str(text).replace("\r", "").split("\n"):
|
|
line = _PaintedLine(line)
|
|
line.width = width
|
|
lines.append(line)
|
|
_cli()._OUTPUT_HISTORY.extend(lines)
|
|
|
|
|
|
_ANSI_SEQUENCE_RE = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)|[@-Z\\-_])")
|
|
|
|
|
|
def _ansi_drop_cells(line: str, cells: int) -> str:
|
|
"""``line`` without its first ``cells`` visible cells; escape sequences are kept for their styling."""
|
|
from prompt_toolkit.utils import get_cwidth
|
|
out, pos = [], 0
|
|
for match in [*_ANSI_SEQUENCE_RE.finditer(line), None]:
|
|
end = match.start() if match else len(line)
|
|
for ch in line[pos:end]:
|
|
if cells > 0:
|
|
cells -= get_cwidth(ch)
|
|
else:
|
|
out.append(ch)
|
|
if match:
|
|
out.append(match.group())
|
|
pos = match.end()
|
|
return "".join(out)
|
|
|
|
|
|
def _terminal_reflows() -> bool:
|
|
"""Whether the terminal re-wraps the rows it shows when its width changes.
|
|
|
|
Most do (tmux, VTE, kitty, iTerm2, Terminal.app, WezTerm, Alacritty, Windows Terminal):
|
|
a shrink pushes the rows that grew into scrollback. xterm, GNU screen and the Linux
|
|
console keep every row in place, truncated. Unknown terminals count as reflowing.
|
|
"""
|
|
env = os.environ
|
|
if env.get("TMUX"):
|
|
return True
|
|
if env.get("STY") or env.get("XTERM_VERSION"):
|
|
return False
|
|
return env.get("TERM", "") != "linux"
|
|
|
|
|
|
def _output_tail_fitting(lines: list[str], max_rows: int, columns: int, painted: bool = True) -> list[str]:
|
|
"""Newest ``lines`` filling ``max_rows`` rows, soft-wrapped at ``columns`` — or, with
|
|
``painted``, at the width each was painted at, as a terminal that does not reflow still
|
|
shows it. The oldest one may only partly fit: its bottom rows are kept, the rows above
|
|
them are already in scrollback."""
|
|
from prompt_toolkit.formatted_text import ANSI, fragment_list_width, to_formatted_text
|
|
kept, used = [], 0
|
|
for line in reversed(lines):
|
|
if used >= max_rows:
|
|
break
|
|
cols = (getattr(line, "width", None) if painted else None) or columns
|
|
width = fragment_list_width(to_formatted_text(ANSI(line)))
|
|
height = max(1, -(-width // cols)) if cols > 0 else 1
|
|
if used + height > max_rows:
|
|
kept.append(_ansi_drop_cells(line, (height - (max_rows - used)) * cols))
|
|
break
|
|
used += height
|
|
kept.append(line)
|
|
kept.reverse()
|
|
return kept
|
|
|
|
|
|
def _pt_print_ansi(text: str) -> None:
|
|
"""``_pt_print(ANSI(text))``, falling back to ``print`` when stdout is not a real console."""
|
|
from cli import _PT_ANSI, _pt_print
|
|
try:
|
|
_pt_print(_PT_ANSI(text))
|
|
except Exception:
|
|
# NoConsoleScreenBufferError (Windows) / OSError when stdout is e.g. a worker log file.
|
|
with suppress(Exception):
|
|
print(text)
|
|
|
|
|
|
_HELD_PAINTS: list | None = None
|
|
_HELD_PAINTS_LOCK = threading.Lock()
|
|
|
|
|
|
def _hold_paints() -> None:
|
|
"""Hold ``_cprint`` paints until ``_release_paints``, while a resize awaits its recovery.
|
|
|
|
A paint erases the prompt chrome from prompt_toolkit's cursor, which is stale once the
|
|
terminal re-wrapped the chrome to its new width: rows of the old chrome would stay in the
|
|
transcript, where the replay cannot account for them (#95375).
|
|
"""
|
|
global _HELD_PAINTS
|
|
with _HELD_PAINTS_LOCK:
|
|
if _HELD_PAINTS is None:
|
|
_HELD_PAINTS = []
|
|
|
|
|
|
def _release_paints() -> None:
|
|
"""Paint, in order, what ``_hold_paints`` held."""
|
|
global _HELD_PAINTS
|
|
with _HELD_PAINTS_LOCK:
|
|
held, _HELD_PAINTS = _HELD_PAINTS or [], None
|
|
for paint in held:
|
|
with suppress(Exception):
|
|
paint()
|
|
|
|
|
|
def _paint_held(paint) -> bool:
|
|
with _HELD_PAINTS_LOCK:
|
|
if _HELD_PAINTS is None:
|
|
return False
|
|
_HELD_PAINTS.append(paint)
|
|
return True
|
|
|
|
|
|
def _cprint(text: str):
|
|
"""Print ANSI text through prompt_toolkit's renderer (patch_stdout swallows raw ANSI).
|
|
|
|
From a background thread while an Application runs, a direct print races the input
|
|
redraw and gets buried, so those go through ``run_in_terminal`` via ``call_soon_threadsafe``.
|
|
"""
|
|
from cli import _PT_ANSI, _output_history_recording, _pt_print, _pt_print_ansi, _record_output_history
|
|
recording = _output_history_recording()
|
|
|
|
def _painted(paint):
|
|
# Recorded when painted, not when requested: a redraw replaying the history must
|
|
# neither print rows still queued for the loop nor size them at a stale width.
|
|
def _paint():
|
|
if recording:
|
|
_record_output_history(text, force=True)
|
|
paint()
|
|
return _paint
|
|
paint_pt = _painted(lambda: _pt_print(_PT_ANSI(text)))
|
|
paint_fallback = _painted(lambda: _pt_print_ansi(text))
|
|
|
|
try:
|
|
from prompt_toolkit.application import get_app_or_none, run_in_terminal
|
|
except Exception:
|
|
paint_pt()
|
|
return
|
|
|
|
try:
|
|
app = get_app_or_none()
|
|
except Exception:
|
|
app = None
|
|
|
|
if app is None or not getattr(app, "_is_running", False):
|
|
_release_paints()
|
|
paint_fallback()
|
|
return
|
|
|
|
import asyncio as _asyncio
|
|
|
|
try:
|
|
loop = app.loop # type: ignore[attr-defined]
|
|
except Exception:
|
|
loop = None
|
|
try:
|
|
# get_running_loop(): get_event_loop() warns from threads with no current loop.
|
|
# Use get_running_loop() instead of get_event_loop() to avoid the DeprecationWarning /
|
|
# RuntimeWarning emitted by Python 3.10+ when get_event_loop() is called from a thread that has no
|
|
# current event loop set (e.g. the process_loop background thread). Fixes #19285.
|
|
current_loop = _asyncio.get_running_loop()
|
|
except Exception:
|
|
current_loop = None
|
|
if loop is None or (current_loop is loop and loop.is_running()):
|
|
if not _paint_held(paint_pt):
|
|
paint_pt()
|
|
return
|
|
|
|
def _schedule():
|
|
# run_in_terminal() returns an awaitable (pt >= 3.0) that must be scheduled or the
|
|
# output is dropped, or None (mocks / older pt) when it already ran synchronously.
|
|
# Never fall back to a bare print on error: the sync path already printed.
|
|
if _paint_held(paint_pt):
|
|
return
|
|
with suppress(Exception):
|
|
import inspect as _inspect
|
|
coro = run_in_terminal(paint_pt)
|
|
if coro is not None and (_inspect.isawaitable(coro) or _inspect.iscoroutine(coro)):
|
|
_asyncio.ensure_future(coro)
|
|
|
|
try:
|
|
loop.call_soon_threadsafe(_schedule)
|
|
except Exception:
|
|
paint_fallback()
|
|
|
|
|
|
def _prepend_note_to_message(message, note: str):
|
|
"""Prepend a one-shot note to a user message (str, or content-part list when an image is attached).
|
|
|
|
For lists the note is folded into the first text part or inserted as a leading one.
|
|
Unknown shapes are returned unchanged.
|
|
"""
|
|
note = str(note or "").strip()
|
|
if not note:
|
|
return message
|
|
if isinstance(message, str):
|
|
return f"{note}\n\n{message}" if message else note
|
|
if isinstance(message, list):
|
|
parts = list(message)
|
|
for i, part in enumerate(parts):
|
|
if isinstance(part, dict) and part.get("type") == "text":
|
|
text = part.get("text", "")
|
|
parts[i] = {**part, "text": f"{note}\n\n{text}" if text else note}
|
|
return parts
|
|
return [{"type": "text", "text": note}, *parts]
|
|
return message
|
|
|
|
|
|
def _pt_app_is_running() -> bool:
|
|
"""Whether a prompt_toolkit Application currently owns the live terminal."""
|
|
try:
|
|
from prompt_toolkit.application import get_app_or_none
|
|
app = get_app_or_none()
|
|
except Exception:
|
|
return False
|
|
return app is not None and bool(getattr(app, "_is_running", False))
|
|
|
|
|
|
def _cli_visible_print(text: str = "") -> None:
|
|
"""``print`` unless a prompt_toolkit Application owns the terminal (patch_stdout swallows bare prints)."""
|
|
from cli import _cprint, _pt_app_is_running
|
|
if _pt_app_is_running():
|
|
_cprint(text)
|
|
else:
|
|
print(text)
|
|
|
|
|
|
class ChatConsole:
|
|
"""Rich Console drop-in routing rendered ANSI through ``_cprint`` so colors survive patch_stdout."""
|
|
|
|
def __init__(self):
|
|
from io import StringIO
|
|
self._buffer = StringIO()
|
|
self._inner = Console(file=self._buffer, force_terminal=True, color_system="truecolor", highlight=False)
|
|
|
|
def print(self, *args, **kwargs):
|
|
from cli import _OSC_ESCAPE_RE, _cprint
|
|
self._buffer.seek(0)
|
|
self._buffer.truncate()
|
|
self._inner.width = shutil.get_terminal_size((80, 24)).columns
|
|
self._inner.print(*args, **kwargs)
|
|
for line in _OSC_ESCAPE_RE.sub("", self._buffer.getvalue()).rstrip("\n").split("\n"):
|
|
_cprint(line)
|
|
|
|
@contextmanager
|
|
def status(self, *_args, **_kwargs):
|
|
"""No-op ``console.status`` so slash helpers don't duplicate ``_busy_command()``'s indicator."""
|
|
yield self
|
|
|
|
|
|
def _build_compact_banner() -> str:
|
|
"""Build a compact banner that fits the current terminal width."""
|
|
try:
|
|
from hermes_cli.skin_engine import get_active_skin
|
|
_skin = get_active_skin()
|
|
except Exception:
|
|
_skin = None
|
|
|
|
def _color(key, default):
|
|
return _skin.get_color(key, default) if _skin else default
|
|
|
|
border_color = _color("banner_border", "#FFD700")
|
|
title_color = _color("banner_title", "#FFBF00")
|
|
dim_color = _color("banner_dim", "#B8860B")
|
|
|
|
if (getattr(_skin, "name", "default") if _skin else "default") == "default":
|
|
tiny_line = "☤ NOUS HERMES"
|
|
else:
|
|
tiny_line = _skin.get_branding("agent_name", "Hermes Agent") if _skin else "Hermes Agent"
|
|
line1 = f"{tiny_line} - AI Agent Framework"
|
|
|
|
if os.environ.get("HERMES_FAST_STARTUP_BANNER") == "1":
|
|
from hermes_cli import __release_date__ as _release_date
|
|
from hermes_cli import __version__ as _version
|
|
|
|
version_line = f"Hermes Agent v{_version} ({_release_date})"
|
|
else:
|
|
version_line = format_banner_version_label()
|
|
|
|
w = min(shutil.get_terminal_size().columns - 2, 88)
|
|
if w < 30:
|
|
return f"\n[{title_color}]{tiny_line}[/] [dim {dim_color}]- Nous Research[/]\n"
|
|
|
|
inner = w - 2 # inside the box border
|
|
bar = "═" * w
|
|
content_width = inner - 2
|
|
|
|
line1 = line1[:content_width].ljust(content_width)
|
|
line2 = version_line[:content_width].ljust(content_width)
|
|
|
|
return (
|
|
f"\n[bold {border_color}]╔{bar}╗[/]\n"
|
|
f"[bold {border_color}]║[/] [{title_color}]{line1}[/] [bold {border_color}]║[/]\n"
|
|
f"[bold {border_color}]║[/] [dim {dim_color}]{line2}[/] [bold {border_color}]║[/]\n"
|
|
f"[bold {border_color}]╚{bar}╝[/]\n"
|
|
)
|
|
|
|
|
|
def _panel_box_width(title: str, content_lines: list[str], min_width: int = 46, max_width: int = 76) -> int:
|
|
"""Stable TUI panel width wide enough for the title and content (incl. borders)."""
|
|
term_cols = shutil.get_terminal_size((100, 20)).columns
|
|
longest = max([len(title)] + [len(line) for line in content_lines] + [min_width - 4])
|
|
inner = min(max(longest + 4, min_width - 2), max_width - 2, max(24, term_cols - 6))
|
|
return inner + 2 # leading/trailing space inside the borders
|
|
|
|
|
|
def _wrap_panel_text(text: str, width: int, subsequent_indent: str = "", *, keep_ws: bool = False) -> list[str]:
|
|
"""Wrap panel text; ``keep_ws`` preserves whitespace (command/detail previews)."""
|
|
kw = dict(replace_whitespace=False, drop_whitespace=False) if keep_ws else dict(break_long_words=False, break_on_hyphens=False)
|
|
wrapped = textwrap.wrap(text, width=max(8, width), subsequent_indent=subsequent_indent, **kw)
|
|
return wrapped or [""]
|
|
|
|
|
|
_wrap_panel_text_keep_ws = functools.partial(_wrap_panel_text, keep_ws=True)
|
|
|
|
|
|
def _append_panel_line(lines, border_style: str, content_style: str, text: str, box_width: int) -> None:
|
|
lines.extend(((border_style, "│ "), (content_style, text.ljust(max(0, box_width - 2))), (border_style, " │\n")))
|
|
|
|
|
|
def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
|
|
lines.append((border_style, "│" + (" " * box_width) + "│\n"))
|