1300 lines
50 KiB
Python
1300 lines
50 KiB
Python
"""CLI presentation -- spinner, kawaii faces, tool preview formatting.
|
||
|
||
Pure display functions with no AIAgent dependency; used for CLI feedback.
|
||
"""
|
||
|
||
import logging
|
||
import os
|
||
import re
|
||
import sys
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
from difflib import unified_diff
|
||
from pathlib import Path
|
||
from typing import Any, Iterator
|
||
from urllib.parse import urlsplit
|
||
|
||
from utils import safe_json_loads
|
||
from agent.redact import redact_sensitive_text
|
||
from agent.tool_result_classification import file_mutation_result_landed
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_ANSI_RESET = "\033[0m"
|
||
_MAX_INLINE_DIFF_FILES = 6
|
||
_MAX_INLINE_DIFF_LINES = 80
|
||
|
||
|
||
def _display_url(value: Any) -> str:
|
||
"""Extract a display-only URL without assuming model argument types."""
|
||
if isinstance(value, dict):
|
||
value = value.get("url") or value.get("href")
|
||
return value.strip() if isinstance(value, str) else ""
|
||
|
||
|
||
def _hex_rgb(h: str) -> tuple[int, int, int]:
|
||
return int(h[1:3], 16), int(h[3:5], 16), int(h[5:7], 16)
|
||
|
||
|
||
def _get_skin():
|
||
"""Get the active skin config, or None if not available (lazy import avoids cycles)."""
|
||
try:
|
||
from hermes_cli.skin_engine import get_active_skin
|
||
return get_active_skin()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
# Diff colors resolve lazily from the skin engine (light/dark aware) and are
|
||
# cached after the first resolution.
|
||
_diff_colors_cached: dict[str, str] | None = None
|
||
|
||
|
||
def _diff_ansi() -> dict[str, str]:
|
||
"""Return ANSI escapes for diff display, resolved from the active skin."""
|
||
global _diff_colors_cached
|
||
if _diff_colors_cached is not None:
|
||
return _diff_colors_cached
|
||
|
||
def _fg(r: int, g: int, b: int) -> str:
|
||
return f"\033[38;2;{r};{g};{b}m"
|
||
|
||
# Defaults that work on dark terminals.
|
||
colors = {
|
||
"dim": _fg(150, 150, 150), "file": _fg(180, 160, 255), "hunk": _fg(120, 120, 140),
|
||
"minus": "\033[38;2;255;255;255;48;2;120;20;20m", "plus": "\033[38;2;255;255;255;48;2;20;90;20m",
|
||
}
|
||
try:
|
||
skin = _get_skin()
|
||
|
||
def _skin_fg(key: str, fallback_rgb: tuple[int, int, int]) -> str:
|
||
h = skin.get_color(key, "")
|
||
return _fg(*(_hex_rgb(h) if h and len(h) == 7 and h[0] == "#" else fallback_rgb))
|
||
|
||
colors["dim"] = _skin_fg("banner_dim", (150, 150, 150))
|
||
colors["file"] = _skin_fg("session_label", (180, 160, 255))
|
||
colors["hunk"] = _skin_fg("session_border", (120, 120, 140))
|
||
# minus/plus use dark-tinted backgrounds derived from ui_error/ui_ok.
|
||
err_h = skin.get_color("ui_error", "#ef5350")
|
||
ok_h = skin.get_color("ui_ok", "#4caf50")
|
||
if err_h and len(err_h) == 7:
|
||
er, eg, eb = _hex_rgb(err_h)
|
||
colors["minus"] = f"\033[38;2;255;255;255;48;2;{max(er//2,20)};{max(eg//4,10)};{max(eb//4,10)}m"
|
||
if ok_h and len(ok_h) == 7:
|
||
or_, og, ob = _hex_rgb(ok_h)
|
||
colors["plus"] = f"\033[38;2;255;255;255;48;2;{max(or_//4,10)};{max(og//2,20)};{max(ob//4,10)}m"
|
||
except Exception:
|
||
pass
|
||
_diff_colors_cached = colors
|
||
return colors
|
||
|
||
|
||
@dataclass
|
||
class LocalEditSnapshot:
|
||
"""Pre-tool filesystem snapshot used to render diffs locally after writes."""
|
||
paths: list[Path] = field(default_factory=list)
|
||
before: dict[str, str | None] = field(default_factory=dict)
|
||
|
||
|
||
# Configurable tool preview length; set once at startup from display.tool_preview_length.
|
||
_tool_preview_max_len: int = 0 # 0 = unlimited
|
||
|
||
|
||
def set_tool_preview_max_len(n: int) -> None:
|
||
"""Set the global max length for tool call previews. 0 = no limit."""
|
||
global _tool_preview_max_len
|
||
_tool_preview_max_len = max(int(n), 0) if n else 0
|
||
|
||
|
||
def get_tool_preview_max_len() -> int:
|
||
"""Return the configured max preview length (0 = unlimited)."""
|
||
return _tool_preview_max_len
|
||
|
||
|
||
def get_skin_tool_prefix() -> str:
|
||
"""Get tool output prefix character from active skin."""
|
||
skin = _get_skin()
|
||
return skin.tool_prefix if skin else "┊"
|
||
|
||
|
||
def get_tool_emoji(tool_name: str, default: str = "⚡") -> str:
|
||
"""Display emoji for a tool: skin ``tool_emojis`` override, then registry, then *default*."""
|
||
skin = _get_skin()
|
||
if skin and skin.tool_emojis:
|
||
override = skin.tool_emojis.get(tool_name)
|
||
if override:
|
||
return override
|
||
try:
|
||
from tools.registry import registry
|
||
emoji = registry.get_emoji(tool_name, default="")
|
||
if emoji:
|
||
return emoji
|
||
except Exception:
|
||
pass
|
||
return default
|
||
|
||
|
||
# =========================================================================
|
||
# Tool preview (one-line summary of a tool call's primary argument)
|
||
# =========================================================================
|
||
|
||
def _oneline(text: str) -> str:
|
||
"""Collapse whitespace (including newlines) to single spaces."""
|
||
return " ".join(text.split())
|
||
|
||
|
||
def _tail_trunc(text: str, limit: int) -> str:
|
||
"""Tail-truncate to ``limit`` chars with ``...`` (0 = unlimited; no guard for limit <= 3)."""
|
||
return text[:limit - 3] + "..." if limit > 0 and len(text) > limit else text
|
||
|
||
|
||
def _truncate_preview(text: str, max_len: int | None) -> str:
|
||
if max_len and max_len > 0 and len(text) > max_len:
|
||
return "." * max_len if max_len <= 3 else text[:max_len - 3] + "..."
|
||
return text
|
||
|
||
|
||
def _clip(text: str, n: int) -> str:
|
||
"""``text[:n]`` plus ``...`` when longer (used inside quoted previews)."""
|
||
return f"{text[:n]}{'...' if len(text) > n else ''}"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ToolPreview:
|
||
"""A compact tool preview plus presentation facts lost to truncation."""
|
||
|
||
text: str
|
||
truncated: bool = False
|
||
url: str | None = None
|
||
|
||
|
||
# ── Shell command summarisation ──────────────────────────────────────────
|
||
|
||
_SHELL_SILENT_HEADS = {"cd", "pushd", "popd", "export", "set", "unset", "source", ".", "true", "false", ":"}
|
||
_SHELL_PIPE_TAIL_HEADS = {"head", "tail", "wc", "sort", "uniq"}
|
||
|
||
|
||
def _shell_basename(head: str) -> str:
|
||
return head.rsplit("/", 1)[-1] if head else ""
|
||
|
||
|
||
def _scan_quoted(text: str) -> Iterator[tuple[int, str, bool]]:
|
||
"""Yield ``(index, char, inside_quotes)``; a quote closes unless backslash-escaped."""
|
||
quote: str | None = None
|
||
for i, ch in enumerate(text):
|
||
if quote:
|
||
yield i, ch, True
|
||
if ch == quote and (i == 0 or text[i - 1] != "\\"):
|
||
quote = None
|
||
elif ch in {"'", '"'}:
|
||
quote = ch
|
||
yield i, ch, True
|
||
else:
|
||
yield i, ch, False
|
||
|
||
|
||
def _split_shell_words(segment: str) -> list[str]:
|
||
words: list[str] = []
|
||
buf: list[str] = []
|
||
for _, ch, quoted in _scan_quoted(segment):
|
||
if not quoted and ch.isspace():
|
||
if buf:
|
||
words.append("".join(buf))
|
||
buf = []
|
||
else:
|
||
buf.append(ch)
|
||
if buf:
|
||
words.append("".join(buf))
|
||
return words
|
||
|
||
|
||
def _strip_shell_pipe_tail(segment: str) -> str:
|
||
words = _split_shell_words(segment)
|
||
out: list[str] = []
|
||
for i, word in enumerate(words):
|
||
if word == "|" and _shell_basename(words[i + 1] if i + 1 < len(words) else "") in _SHELL_PIPE_TAIL_HEADS:
|
||
break
|
||
out.append(word)
|
||
return " ".join(out).strip()
|
||
|
||
|
||
def _split_shell_compound(command: str) -> list[str]:
|
||
"""Split on unquoted ``&&`` / ``||`` / ``;`` / newline, dropping pipe tails per segment."""
|
||
segments: list[str] = []
|
||
buf: list[str] = []
|
||
skip = False
|
||
|
||
def _flush() -> None:
|
||
segment = _strip_shell_pipe_tail("".join(buf).strip())
|
||
if segment:
|
||
segments.append(segment)
|
||
buf.clear()
|
||
|
||
for i, ch, quoted in _scan_quoted(command):
|
||
if skip:
|
||
skip = False
|
||
continue
|
||
if not quoted and (command.startswith("&&", i) or command.startswith("||", i)):
|
||
_flush()
|
||
skip = True
|
||
continue
|
||
if not quoted and ch in {";", "\n"}:
|
||
_flush()
|
||
continue
|
||
buf.append(ch)
|
||
_flush()
|
||
return segments
|
||
|
||
|
||
def _shell_head_word(segment: str) -> str:
|
||
words = _split_shell_words(segment)
|
||
index = 0
|
||
while index < len(words) and re.match(r"^[A-Za-z_]\w*=", words[index]):
|
||
index += 1
|
||
return _shell_basename(words[index] if index < len(words) else "")
|
||
|
||
|
||
def _clean_shell_segment(segment: str) -> str:
|
||
"""Drop redirections (``> file``, ``2>&1``) from a segment."""
|
||
words = _split_shell_words(segment)
|
||
out: list[str] = []
|
||
i = 0
|
||
while i < len(words):
|
||
word = words[i]
|
||
if re.match(r"^\d*(?:>>?|<)$", word):
|
||
i += 2
|
||
continue
|
||
if re.match(r"^\d*(?:>&|<&)\d+$", word) or re.match(r"^\d*>&\d+$", word):
|
||
i += 1
|
||
continue
|
||
out.append(word)
|
||
i += 1
|
||
return " ".join(out).strip()
|
||
|
||
|
||
def _is_shell_boundary_echo(segment: str) -> bool:
|
||
words = _split_shell_words(segment)
|
||
if _shell_basename(words[0] if words else "") != "echo":
|
||
return False
|
||
return bool(re.search(r"-{2,}|_exit=|(?:^|\s|=)\$[?{]|PIPESTATUS", " ".join(words[1:])))
|
||
|
||
|
||
def summarize_shell_command(command: str) -> str:
|
||
"""Compact shell wrapper/plumbing for display while preserving raw command elsewhere."""
|
||
original = _oneline(command)
|
||
if not original:
|
||
return ""
|
||
segments = _split_shell_compound(original)
|
||
if len(segments) <= 1:
|
||
return _clean_shell_segment(segments[0] if segments else original) or original
|
||
core: list[str] = []
|
||
for segment in segments:
|
||
cleaned = _clean_shell_segment(segment)
|
||
if cleaned and _shell_head_word(cleaned) not in _SHELL_SILENT_HEADS and not _is_shell_boundary_echo(cleaned):
|
||
core.append(cleaned)
|
||
if not core:
|
||
return original
|
||
if len(core) == 1:
|
||
return core[0]
|
||
count = len(core) - 1
|
||
return f"{core[0]} + {count} {'command' if count == 1 else 'commands'}"
|
||
|
||
|
||
def _read_file_line_label(args: dict) -> str:
|
||
offset = args.get("offset")
|
||
limit = args.get("limit")
|
||
if not isinstance(offset, int) or offset <= 0:
|
||
return ""
|
||
if not isinstance(limit, int) or limit <= 1:
|
||
return f"L{offset}"
|
||
return f"L{offset}-{offset + limit - 1}"
|
||
|
||
|
||
def redact_browser_typed_text_for_display(value: Any, typed_text: Any) -> Any:
|
||
"""Replace every occurrence of a secret-looking browser_type value with its redacted form.
|
||
|
||
Backends echo the attempted input in error strings/fallback metadata, so the raw
|
||
value is swapped before the result reaches logs, callbacks, the model, or chat
|
||
history. Normal typed text matches no secret pattern and passes through unchanged.
|
||
Redaction is forced regardless of ``security.redact_secrets``: a typed credential
|
||
leaking into chat history is a security boundary, not log hygiene.
|
||
"""
|
||
if typed_text is None:
|
||
return value
|
||
needle = str(typed_text)
|
||
if needle == "":
|
||
return value
|
||
redacted = redact_sensitive_text(needle, force=True)
|
||
if redacted == needle:
|
||
return value
|
||
if isinstance(value, str):
|
||
return value.replace(needle, redacted)
|
||
if isinstance(value, dict):
|
||
return {key: redact_browser_typed_text_for_display(item, typed_text) for key, item in value.items()}
|
||
if isinstance(value, list):
|
||
return [redact_browser_typed_text_for_display(item, typed_text) for item in value]
|
||
if isinstance(value, tuple):
|
||
return tuple(redact_browser_typed_text_for_display(item, typed_text) for item in value)
|
||
return value
|
||
|
||
|
||
def redact_tool_args_for_display(tool_name: str, args: dict | None) -> dict | None:
|
||
"""Return a copy of tool args safe for logs/progress UI (masks ``browser_type`` secrets)."""
|
||
if not isinstance(args, dict):
|
||
return args
|
||
if tool_name == "browser_type" and isinstance(args.get("text"), str):
|
||
return {**args, "text": redact_sensitive_text(args["text"], force=True)}
|
||
return args
|
||
|
||
|
||
def _delegate_task_goals(tasks: Any, *, per_goal_len: int) -> list[str]:
|
||
"""One truncated goal string per dict task (``?`` when missing)."""
|
||
if not isinstance(tasks, list):
|
||
return []
|
||
goals: list[str] = []
|
||
for task in tasks:
|
||
if isinstance(task, dict):
|
||
raw_goal = task.get("goal")
|
||
goal = "?" if raw_goal is None else _oneline(str(raw_goal))
|
||
goals.append(_truncate_preview(goal or "?", per_goal_len))
|
||
return goals
|
||
|
||
|
||
def _browser_exec_step_label(args: dict, max_chars: int = 80) -> str | None:
|
||
"""User-friendly step label from browser_exec code's leading comment."""
|
||
code = str(args.get("code", "") or "").strip()
|
||
if not code:
|
||
return None
|
||
first = code.split("\n", 1)[0].strip()
|
||
if not first.startswith("#"):
|
||
return None
|
||
label = first.lstrip("#").strip()
|
||
if not label:
|
||
return None
|
||
if len(label) > max_chars:
|
||
label = label[: max_chars - 1] + "…"
|
||
return label
|
||
|
||
|
||
_PRIMARY_ARGS = {
|
||
"terminal": "command", "web_search": "query", "web_extract": "urls",
|
||
"read_file": "path", "write_file": "path", "patch": "path",
|
||
"search_files": "pattern", "browser_navigate": "url",
|
||
"browser_click": "ref", "browser_type": "text",
|
||
"image_generate": "prompt", "text_to_speech": "text",
|
||
"vision_analyze": "question",
|
||
"skill_view": "name", "skills_list": "category",
|
||
"cronjob_manage": "action",
|
||
"execute_code": "code", "browser_exec": "code", "delegate_task": "goal",
|
||
"clarify": "question", "skill_manage": "name",
|
||
}
|
||
_FALLBACK_PREVIEW_KEYS = ("query", "text", "command", "path", "name", "prompt", "code", "goal")
|
||
|
||
|
||
def _delegate_action_preview(args: dict) -> str | None:
|
||
"""Shared ``list/steer/stop <id>`` preview for delegate_task, or None for spawn calls."""
|
||
action = str(args.get("action") or "").strip().lower()
|
||
if action in ("list", "steer", "stop"):
|
||
return f"{action} {str(args.get('subagent_id') or '').strip()}".strip()
|
||
return None
|
||
|
||
|
||
def _preview_browser_exec(args: dict, max_len: int) -> str | None:
|
||
label = _browser_exec_step_label(args)
|
||
if label is not None:
|
||
return _truncate_preview(label, max_len)
|
||
return _truncate_preview(_oneline(str(args.get("code", "") or "")), max_len) or None
|
||
|
||
|
||
def _preview_delegate_task(args: dict, max_len: int) -> str | None:
|
||
action_preview = _delegate_action_preview(args)
|
||
if action_preview is not None:
|
||
return _truncate_preview(action_preview, max_len)
|
||
tasks = args.get("tasks")
|
||
if tasks and isinstance(tasks, list):
|
||
goals = _delegate_task_goals(tasks, per_goal_len=40)
|
||
preview = f"{len(goals)} tasks: " + " | ".join(goals) if goals else f"{len(tasks)} parallel tasks"
|
||
return _truncate_preview(preview, max_len)
|
||
goal = args.get("goal", "")
|
||
if goal is None:
|
||
return None
|
||
return _truncate_preview(_oneline(str(goal)), max_len) or None
|
||
|
||
|
||
def _preview_process_manage(args: dict, _max_len: int) -> str | None:
|
||
action = args.get("action", "")
|
||
sid = args.get("session_id", "")
|
||
data = args.get("data", "")
|
||
timeout_val = args.get("timeout")
|
||
parts = [str(action) if action else ""]
|
||
if sid:
|
||
parts.append(str(sid)[:16])
|
||
if data:
|
||
parts.append(f'"{_oneline(str(data)[:20])}"')
|
||
if timeout_val and action == "wait":
|
||
parts.append(f"{timeout_val}s")
|
||
parts = [p for p in parts if p]
|
||
return " ".join(parts) if parts else None
|
||
|
||
|
||
def _preview_todo_list(args: dict, _max_len: int) -> str:
|
||
todos_arg = args.get("todos")
|
||
if todos_arg is None:
|
||
return "reading task list"
|
||
return f"{'updating' if args.get('merge', False) else 'planning'} {len(todos_arg)} task(s)"
|
||
|
||
|
||
def _preview_shell(key: str):
|
||
def _build(args: dict, max_len: int) -> str | None:
|
||
command = args.get(key)
|
||
if command is None:
|
||
return None
|
||
return _truncate_preview(summarize_shell_command(str(command)), max_len) or None
|
||
return _build
|
||
|
||
|
||
def _preview_read_file(args: dict, max_len: int) -> str | None:
|
||
path = args.get("path") or args.get("file") or args.get("filepath")
|
||
if path is None:
|
||
return None
|
||
label = Path(str(path).replace("\\", "/")).name or str(path)
|
||
return _truncate_preview(f"{label} {_read_file_line_label(args)}".strip(), max_len) or None
|
||
|
||
|
||
def _preview_memory(args: dict, _max_len: int) -> str:
|
||
action = args.get("action", "")
|
||
target = args.get("target", "")
|
||
if action == "add":
|
||
return f"+{target}: \"{_clip(_oneline(args.get('content', '')), 25)}\""
|
||
if action in ("replace", "remove"):
|
||
old = _oneline(args.get("old_text") or "") or "<missing old_text>"
|
||
return f"{'~' if action == 'replace' else '-'}{target}: \"{old[:20]}\""
|
||
return action
|
||
|
||
|
||
def _preview_send_message(args: dict, _max_len: int) -> str:
|
||
msg = _oneline(args.get("message", ""))
|
||
if len(msg) > 20:
|
||
msg = msg[:17] + "..."
|
||
return f"to {args.get('target', '?')}: \"{msg}\""
|
||
|
||
|
||
def _preview_skill_view(args: dict, max_len: int) -> str | None:
|
||
name = _oneline(str(args.get("name") or ""))
|
||
file_path = args.get("file_path")
|
||
if file_path:
|
||
file_path = _oneline(str(file_path))
|
||
return _truncate_preview(f"{name} → {file_path}" if name else file_path, max_len) or None
|
||
return _truncate_preview(name, max_len) or None
|
||
|
||
|
||
# Tool-specific preview builders: f(args, max_len) -> preview. Tools not listed
|
||
# fall through to the primary-argument lookup in build_tool_preview.
|
||
_PREVIEW_BUILDERS = {
|
||
"browser_exec": _preview_browser_exec,
|
||
"delegate_task": _preview_delegate_task,
|
||
"process_manage": _preview_process_manage,
|
||
"todo_list": _preview_todo_list,
|
||
"terminal": _preview_shell("command"),
|
||
"execute_code": _preview_shell("code"),
|
||
"read_file": _preview_read_file,
|
||
"session_search": lambda args, _m: f"recall: \"{_clip(_oneline(args.get('query', '')), 25)}\"",
|
||
"memory": _preview_memory,
|
||
"send_message": _preview_send_message,
|
||
"skill_view": _preview_skill_view,
|
||
}
|
||
|
||
|
||
def build_tool_preview(tool_name: str, args: dict, max_len: int | None = None) -> str | None:
|
||
"""Build a short preview of a tool call's primary argument for display.
|
||
|
||
*max_len* ``None`` defers to the global ``_tool_preview_max_len``; ``0`` means unlimited.
|
||
"""
|
||
if max_len is None:
|
||
max_len = _tool_preview_max_len
|
||
if not args:
|
||
return None
|
||
args = redact_tool_args_for_display(tool_name, args) or args
|
||
|
||
builder = _PREVIEW_BUILDERS.get(tool_name)
|
||
if builder is not None:
|
||
return builder(args, max_len)
|
||
|
||
key = _PRIMARY_ARGS.get(tool_name) or next((k for k in _FALLBACK_PREVIEW_KEYS if k in args), None)
|
||
if not key or key not in args:
|
||
return None
|
||
value = args[key]
|
||
if isinstance(value, list):
|
||
value = value[0] if value else ""
|
||
preview = _oneline(str(value))
|
||
return _tail_trunc(preview, max_len) if preview else None
|
||
|
||
|
||
def prepare_tool_preview(
|
||
tool_name: str,
|
||
args: dict | None,
|
||
*,
|
||
fallback: str,
|
||
max_len: int,
|
||
) -> ToolPreview:
|
||
"""Build one canonical compact preview plus explicit truncation/URL metadata.
|
||
|
||
The uncapped preview is rebuilt from the arguments so an upstream display cap
|
||
cannot discard its link target; platforms get truncation/URL facts explicitly
|
||
instead of inferring them from the rendered text.
|
||
"""
|
||
full_text = build_tool_preview(tool_name, args, max_len=0) or fallback
|
||
text = _truncate_preview(full_text, max_len)
|
||
truncated = text != full_text
|
||
url = None
|
||
if truncated:
|
||
candidate = _display_url(full_text)
|
||
try:
|
||
parsed = urlsplit(candidate)
|
||
except ValueError:
|
||
parsed = None
|
||
if parsed and parsed.scheme.lower() in {"http", "https"} and parsed.netloc:
|
||
url = candidate
|
||
return ToolPreview(text=text, truncated=truncated, url=url)
|
||
|
||
|
||
# =========================================================================
|
||
# Friendly tool labels: "web_search <q>" -> "Searching the web for <q>".
|
||
# Curated built-ins only — we know each core tool's semantics so the verb is fixed,
|
||
# not computed; custom/plugin/MCP tools have no entry and fall back to the raw preview.
|
||
# =========================================================================
|
||
|
||
_TOOL_VERBS: dict[str, str] = {
|
||
"web_search": "Searching the web", "web_extract": "Reading",
|
||
"browser_navigate": "Browsing", "browser_click": "Clicking", "browser_type": "Typing",
|
||
"read_file": "Reading", "write_file": "Writing", "patch": "Editing", "search_files": "Searching files",
|
||
"terminal": "Running", "execute_code": "Running code",
|
||
"image_generate": "Generating image", "video_generate": "Generating video",
|
||
"text_to_speech": "Generating speech", "vision_analyze": "Looking at the image",
|
||
"session_search": "Searching past sessions",
|
||
"skill_view": "Reading skill", "skills_list": "Listing skills", "skill_manage": "Updating skill",
|
||
"delegate_task": "Delegating", "cronjob_manage": "Scheduling", "clarify": "Asking",
|
||
"memory": "Updating memory", "todo_list": "Updating tasks",
|
||
}
|
||
# Verbs that read better without the argument preview appended.
|
||
_TOOL_VERBS_NO_PREVIEW: frozenset[str] = frozenset({"skills_list", "session_search"})
|
||
# Verbs joined to the preview with " for " (search-style phrasing).
|
||
_TOOL_VERBS_FOR_CONNECTOR: frozenset[str] = frozenset({"web_search", "search_files"})
|
||
|
||
_friendly_tool_labels: bool = True
|
||
|
||
|
||
def set_friendly_tool_labels(enabled: bool) -> None:
|
||
"""Toggle friendly human-phrased tool labels (display.friendly_tool_labels)."""
|
||
global _friendly_tool_labels
|
||
_friendly_tool_labels = bool(enabled)
|
||
|
||
|
||
def get_tool_verb(tool_name: str) -> str | None:
|
||
"""Friendly verb for a built-in tool, or None (labels disabled / no curated verb).
|
||
|
||
Callers holding a computed preview compose ``f"{verb}{connector}{preview}"``
|
||
themselves via :func:`tool_verb_connector`.
|
||
"""
|
||
return _TOOL_VERBS.get(tool_name) if _friendly_tool_labels else None
|
||
|
||
|
||
def tool_verb_connector(tool_name: str) -> str:
|
||
"""Return the connector between a verb and its preview (" for " or " ")."""
|
||
return " for " if tool_name in _TOOL_VERBS_FOR_CONNECTOR else " "
|
||
|
||
|
||
def verb_drops_preview(tool_name: str) -> bool:
|
||
"""Whether the verb should render alone, without the argument preview."""
|
||
return tool_name in _TOOL_VERBS_NO_PREVIEW
|
||
|
||
|
||
def build_status_phrase(tool_name: str, args: dict | None, max_len: int = 49) -> str | None:
|
||
"""Lowercase "is <verb> <preview>…" phrase for platform status lines (e.g. Slack setStatus).
|
||
|
||
Phrased to follow the bot's display name ("Hermes is running …"). ``args=None``
|
||
gives a verb-only phrase (``display.live_status: verb`` keeps argument previews
|
||
out of shared channels). Returns None for ``_thinking`` or when friendly labels
|
||
are disabled so callers fall back to their static default. Default ``max_len``
|
||
stays under Slack's ~50-char status truncation.
|
||
"""
|
||
if not tool_name or tool_name == "_thinking" or not _friendly_tool_labels:
|
||
return None
|
||
verb = _TOOL_VERBS.get(tool_name)
|
||
phrase = f"is {verb[0].lower()}{verb[1:]}" if verb else f"is using {tool_name}"
|
||
if args and verb and tool_name not in _TOOL_VERBS_NO_PREVIEW:
|
||
preview = build_tool_preview(tool_name, args, max_len=None)
|
||
if preview:
|
||
# Previews can contain newlines (terminal commands); keep the first line.
|
||
phrase = f"{phrase}{tool_verb_connector(tool_name)}{preview.splitlines()[0].strip()}"
|
||
if len(phrase) > max_len - 1:
|
||
return phrase[: max_len - 2].rstrip() + "…"
|
||
return phrase + "…"
|
||
|
||
|
||
def build_tool_label(tool_name: str, args: dict, max_len: int | None = None) -> str | None:
|
||
"""Human-phrased label ("Searching the web for ...") for curated built-ins.
|
||
|
||
Custom/plugin/MCP tools (or labels disabled) get the raw preview, so this is a
|
||
drop-in replacement for :func:`build_tool_preview`.
|
||
"""
|
||
verb = get_tool_verb(tool_name)
|
||
if not verb:
|
||
return build_tool_preview(tool_name, args, max_len=max_len)
|
||
if tool_name in _TOOL_VERBS_NO_PREVIEW:
|
||
return verb
|
||
preview = build_tool_preview(tool_name, args, max_len=max_len)
|
||
return f"{verb}{tool_verb_connector(tool_name)}{preview}" if preview else verb
|
||
|
||
|
||
# =========================================================================
|
||
# Inline diff previews for write actions
|
||
# =========================================================================
|
||
|
||
def _resolved_path(path: str) -> Path:
|
||
"""Resolve a possibly-relative filesystem path against the current cwd."""
|
||
candidate = Path(os.path.expanduser(path))
|
||
return candidate if candidate.is_absolute() else Path.cwd() / candidate
|
||
|
||
|
||
def _snapshot_text(path: Path) -> str | None:
|
||
"""Return UTF-8 file content, or None for missing/unreadable files."""
|
||
try:
|
||
return path.read_text(encoding="utf-8")
|
||
except (FileNotFoundError, IsADirectoryError, UnicodeDecodeError, OSError):
|
||
return None
|
||
|
||
|
||
def _display_diff_path(path: Path) -> str:
|
||
"""Prefer cwd-relative paths in diffs when available."""
|
||
try:
|
||
return str(path.resolve().relative_to(Path.cwd().resolve()))
|
||
except Exception:
|
||
return str(path)
|
||
|
||
|
||
def _resolve_skill_manage_paths(args: dict) -> list[Path]:
|
||
"""Resolve skill_manage write targets to filesystem paths."""
|
||
action = args.get("action")
|
||
name = args.get("name")
|
||
if not action or not name:
|
||
return []
|
||
from tools.skill_manager_tool import _find_skill, _resolve_skill_dir
|
||
|
||
if action == "create":
|
||
return [_resolve_skill_dir(name, args.get("category")) / "SKILL.md"]
|
||
existing = _find_skill(name)
|
||
if not existing:
|
||
return []
|
||
skill_dir = Path(existing["path"])
|
||
file_path = args.get("file_path")
|
||
if action in {"edit", "patch"}:
|
||
return [skill_dir / file_path] if file_path else [skill_dir / "SKILL.md"]
|
||
if action in {"write_file", "remove_file"}:
|
||
return [skill_dir / file_path] if file_path else []
|
||
if action == "delete":
|
||
return [path for path in sorted(skill_dir.rglob("*")) if path.is_file()]
|
||
return []
|
||
|
||
|
||
def _resolve_local_edit_paths(tool_name: str, function_args: dict | None) -> list[Path]:
|
||
"""Resolve local filesystem targets for write-capable tools."""
|
||
if not isinstance(function_args, dict):
|
||
return []
|
||
if tool_name in {"write_file", "patch"}:
|
||
path = function_args.get("path")
|
||
return [_resolved_path(path)] if path else []
|
||
if tool_name == "skill_manage":
|
||
return _resolve_skill_manage_paths(function_args)
|
||
return []
|
||
|
||
|
||
def capture_local_edit_snapshot(tool_name: str, function_args: dict | None) -> LocalEditSnapshot | None:
|
||
"""Capture before-state for local write previews."""
|
||
paths = _resolve_local_edit_paths(tool_name, function_args)
|
||
if not paths:
|
||
return None
|
||
return LocalEditSnapshot(paths=paths, before={str(path): _snapshot_text(path) for path in paths})
|
||
|
||
|
||
def _result_succeeded(result: str | None) -> bool:
|
||
"""Conservatively detect whether a tool result represents success."""
|
||
if not result:
|
||
return False
|
||
data = safe_json_loads(result)
|
||
if not isinstance(data, dict) or data.get("error"):
|
||
return False
|
||
return bool(data.get("success")) if "success" in data else True
|
||
|
||
|
||
def _diff_from_snapshot(snapshot: LocalEditSnapshot | None) -> str | None:
|
||
"""Generate unified diff text from a stored before-state and current files."""
|
||
if not snapshot:
|
||
return None
|
||
chunks: list[str] = []
|
||
for path in snapshot.paths:
|
||
before = snapshot.before.get(str(path))
|
||
after = _snapshot_text(path)
|
||
if before == after:
|
||
continue
|
||
display_path = _display_diff_path(path)
|
||
diff = "".join(unified_diff(
|
||
[] if before is None else before.splitlines(keepends=True),
|
||
[] if after is None else after.splitlines(keepends=True),
|
||
fromfile=f"a/{display_path}", tofile=f"b/{display_path}",
|
||
))
|
||
if diff:
|
||
chunks.append(diff)
|
||
if not chunks:
|
||
return None
|
||
return "".join(chunk if chunk.endswith("\n") else chunk + "\n" for chunk in chunks)
|
||
|
||
|
||
def extract_edit_diff(
|
||
tool_name: str,
|
||
result: str | None,
|
||
*,
|
||
function_args: dict | None = None,
|
||
snapshot: LocalEditSnapshot | None = None,
|
||
) -> str | None:
|
||
"""Extract a unified diff from a file-edit tool result."""
|
||
if tool_name == "patch" and result:
|
||
data = safe_json_loads(result)
|
||
if isinstance(data, dict):
|
||
diff = data.get("diff")
|
||
if isinstance(diff, str) and diff.strip():
|
||
return diff
|
||
if tool_name not in {"write_file", "patch", "skill_manage"} or not _result_succeeded(result):
|
||
return None
|
||
return _diff_from_snapshot(snapshot)
|
||
|
||
|
||
def _emit_inline_diff(diff_text: str, print_fn) -> bool:
|
||
"""Emit rendered diff text through the CLI's prompt_toolkit-safe printer."""
|
||
if print_fn is None or not diff_text:
|
||
return False
|
||
try:
|
||
print_fn(" ┊ review diff")
|
||
for line in diff_text.rstrip("\n").splitlines():
|
||
print_fn(line)
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# Unified-diff line prefix -> diff color key (checked in order; "--- "/"+++ " handled first).
|
||
_DIFF_LINE_COLORS = (("@@", "hunk"), ("-", "minus"), ("+", "plus"), (" ", "dim"))
|
||
|
||
|
||
def _render_inline_unified_diff(diff: str) -> list[str]:
|
||
"""Render unified diff lines in Hermes' inline transcript style."""
|
||
rendered: list[str] = []
|
||
from_file = None
|
||
to_file = None
|
||
for raw_line in diff.splitlines():
|
||
if raw_line.startswith("--- "):
|
||
from_file = raw_line[4:].strip()
|
||
continue
|
||
if raw_line.startswith("+++ "):
|
||
to_file = raw_line[4:].strip()
|
||
if from_file or to_file:
|
||
rendered.append(f"{_diff_ansi()['file']}{from_file or 'a/?'} → {to_file or 'b/?'}{_ANSI_RESET}")
|
||
continue
|
||
color = next((c for prefix, c in _DIFF_LINE_COLORS if raw_line.startswith(prefix)), None)
|
||
if color:
|
||
rendered.append(f"{_diff_ansi()[color]}{raw_line}{_ANSI_RESET}")
|
||
elif raw_line:
|
||
rendered.append(raw_line)
|
||
return rendered
|
||
|
||
|
||
def _split_unified_diff_sections(diff: str) -> list[str]:
|
||
"""Split a unified diff into per-file sections."""
|
||
sections: list[list[str]] = []
|
||
current: list[str] = []
|
||
for line in diff.splitlines():
|
||
if line.startswith("--- ") and current:
|
||
sections.append(current)
|
||
current = [line]
|
||
continue
|
||
current.append(line)
|
||
if current:
|
||
sections.append(current)
|
||
return ["\n".join(section) for section in sections if section]
|
||
|
||
|
||
def _summarize_rendered_diff_sections(
|
||
diff: str,
|
||
*,
|
||
max_files: int = _MAX_INLINE_DIFF_FILES,
|
||
max_lines: int = _MAX_INLINE_DIFF_LINES,
|
||
) -> list[str]:
|
||
"""Render diff sections while capping file count and total line count."""
|
||
sections = _split_unified_diff_sections(diff)
|
||
rendered: list[str] = []
|
||
omitted_files = 0
|
||
omitted_lines = 0
|
||
for idx, section in enumerate(sections):
|
||
section_lines = _render_inline_unified_diff(section)
|
||
remaining_budget = max_lines - len(rendered)
|
||
if idx >= max_files or remaining_budget <= 0:
|
||
omitted_files += 1
|
||
omitted_lines += len(section_lines)
|
||
continue
|
||
if len(section_lines) <= remaining_budget:
|
||
rendered.extend(section_lines)
|
||
continue
|
||
rendered.extend(section_lines[:remaining_budget])
|
||
omitted_lines += len(section_lines) - remaining_budget
|
||
omitted_files += 1 + max(0, len(sections) - idx - 1)
|
||
for leftover in sections[idx + 1:]:
|
||
omitted_lines += len(_render_inline_unified_diff(leftover))
|
||
break
|
||
if omitted_files or omitted_lines:
|
||
summary = f"… omitted {omitted_lines} diff line(s)"
|
||
if omitted_files:
|
||
summary += f" across {omitted_files} additional file(s)/section(s)"
|
||
rendered.append(f"{_diff_ansi()['hunk']}{summary}{_ANSI_RESET}")
|
||
return rendered
|
||
|
||
|
||
def render_edit_diff_with_delta(
|
||
tool_name: str,
|
||
result: str | None,
|
||
*,
|
||
function_args: dict | None = None,
|
||
snapshot: LocalEditSnapshot | None = None,
|
||
print_fn=None,
|
||
) -> bool:
|
||
"""Render an edit diff inline without taking over the terminal UI."""
|
||
diff = extract_edit_diff(tool_name, result, function_args=function_args, snapshot=snapshot)
|
||
if not diff:
|
||
return False
|
||
try:
|
||
rendered_lines = _summarize_rendered_diff_sections(diff)
|
||
except Exception as exc:
|
||
logger.debug("Could not render inline diff: %s", exc)
|
||
return False
|
||
return _emit_inline_diff("\n".join(rendered_lines), print_fn)
|
||
|
||
|
||
# =========================================================================
|
||
# KawaiiSpinner
|
||
# =========================================================================
|
||
|
||
class KawaiiSpinner:
|
||
"""Animated spinner with kawaii faces for CLI feedback during tool execution."""
|
||
|
||
SPINNERS = {
|
||
'dots': ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'],
|
||
'bounce': ['⠁', '⠂', '⠄', '⡀', '⢀', '⠠', '⠐', '⠈'],
|
||
'grow': ['▁', '▂', '▃', '▄', '▅', '▆', '▇', '█', '▇', '▆', '▅', '▄', '▃', '▂'],
|
||
'arrows': ['←', '↖', '↑', '↗', '→', '↘', '↓', '↙'],
|
||
'star': ['✶', '✷', '✸', '✹', '✺', '✹', '✸', '✷'],
|
||
'moon': ['🌑', '🌒', '🌓', '🌔', '🌕', '🌖', '🌗', '🌘'],
|
||
'pulse': ['◜', '◠', '◝', '◞', '◡', '◟'],
|
||
'brain': ['🧠', '💭', '💡', '✨', '💫', '🌟', '💡', '💭'],
|
||
'sparkle': ['⁺', '˚', '*', '✧', '✦', '✧', '*', '˚'],
|
||
}
|
||
|
||
KAWAII_WAITING = [
|
||
"(｡◕‿◕｡)", "(◕‿◕✿)", "٩(◕‿◕｡)۶", "(✿◠‿◠)", "( ˘▽˘)っ",
|
||
"♪(´ε` )", "(◕ᴗ◕✿)", "ヾ(＾∇＾)", "(≧◡≦)", "(★ω★)",
|
||
]
|
||
|
||
KAWAII_THINKING = [
|
||
"(｡•́︿•̀｡)", "(◔_◔)", "(¬‿¬)", "( •_•)>⌐■-■", "(⌐■_■)",
|
||
"(´･_･`)", "◉_◉", "(°ロ°)", "( ˘⌣˘)♡", "ヽ(>∀<☆)☆",
|
||
"٩(๑❛ᴗ❛๑)۶", "(⊙_⊙)", "(¬_¬)", "( ͡° ͜ʖ ͡°)", "ಠ_ಠ",
|
||
]
|
||
|
||
THINKING_VERBS = [
|
||
"pondering", "contemplating", "musing", "cogitating", "ruminating",
|
||
"deliberating", "mulling", "reflecting", "processing", "reasoning",
|
||
"analyzing", "computing", "synthesizing", "formulating", "brainstorming",
|
||
]
|
||
|
||
@staticmethod
|
||
def _skin_spinner_list(key: str, fallback: list) -> list:
|
||
"""Return the active skin's ``spinner[key]`` list, or *fallback* when absent/empty."""
|
||
try:
|
||
skin = _get_skin()
|
||
if skin:
|
||
values = skin.spinner.get(key, [])
|
||
if values:
|
||
return values
|
||
except Exception:
|
||
pass
|
||
return fallback
|
||
|
||
@classmethod
|
||
def get_waiting_faces(cls) -> list:
|
||
return cls._skin_spinner_list("waiting_faces", cls.KAWAII_WAITING)
|
||
|
||
@classmethod
|
||
def get_thinking_faces(cls) -> list:
|
||
return cls._skin_spinner_list("thinking_faces", cls.KAWAII_THINKING)
|
||
|
||
@classmethod
|
||
def get_thinking_verbs(cls) -> list:
|
||
return cls._skin_spinner_list("thinking_verbs", cls.THINKING_VERBS)
|
||
|
||
def __init__(self, message: str = "", spinner_type: str = 'dots', print_fn=None):
|
||
self.message = message
|
||
self.spinner_frames = self.SPINNERS.get(spinner_type, self.SPINNERS['dots'])
|
||
self.running = False
|
||
self.thread = None
|
||
self.frame_idx = 0
|
||
self.start_time = None
|
||
self.last_line_len = 0
|
||
# When set, all output bypasses self._out so silenced agents stay silent.
|
||
self._print_fn = print_fn
|
||
# Capture stdout NOW, before any child redirect_stdout(devnull) replaces it.
|
||
self._out = sys.stdout
|
||
|
||
def _write(self, text: str, end: str = '\n', flush: bool = False):
|
||
"""Write via print_fn when supplied, else to the stdout captured at creation."""
|
||
if self._print_fn is not None:
|
||
try:
|
||
self._print_fn(text)
|
||
except Exception:
|
||
pass
|
||
return
|
||
try:
|
||
self._out.write(text + end)
|
||
if flush:
|
||
self._out.flush()
|
||
except (ValueError, OSError):
|
||
pass
|
||
|
||
@property
|
||
def _is_tty(self) -> bool:
|
||
"""Check if output is a real terminal, safe against closed streams."""
|
||
try:
|
||
return hasattr(self._out, 'isatty') and self._out.isatty()
|
||
except (ValueError, OSError):
|
||
return False
|
||
|
||
def _is_patch_stdout_proxy(self) -> bool:
|
||
"""True when stdout is prompt_toolkit's StdoutProxy.
|
||
|
||
StdoutProxy queues writes and injects newlines around each flush, so the
|
||
\\r overwrite never lands: each frame would land on its own line. The CLI
|
||
drives its own TUI spinner widget in that mode, so we stay silent.
|
||
"""
|
||
try:
|
||
from prompt_toolkit.patch_stdout import StdoutProxy
|
||
return isinstance(self._out, StdoutProxy)
|
||
except ImportError:
|
||
return False
|
||
|
||
def _animate(self):
|
||
# Non-TTY (Docker, systemd, pipe): log once instead of spamming frames.
|
||
if not self._is_tty:
|
||
self._write(f" [tool] {self.message}", flush=True)
|
||
while self.running:
|
||
time.sleep(0.5)
|
||
return
|
||
# Under patch_stdout the \r animation would overdraw the TUI status bar.
|
||
if self._is_patch_stdout_proxy():
|
||
while self.running:
|
||
time.sleep(0.1)
|
||
return
|
||
skin = _get_skin()
|
||
wings = skin.get_spinner_wings() if skin else []
|
||
while self.running:
|
||
if os.getenv("HERMES_SPINNER_PAUSE"):
|
||
time.sleep(0.1)
|
||
continue
|
||
frame = self.spinner_frames[self.frame_idx % len(self.spinner_frames)]
|
||
elapsed = time.time() - self.start_time
|
||
if wings:
|
||
left, right = wings[self.frame_idx % len(wings)]
|
||
line = f" {left} {frame} {self.message} {right} ({elapsed:.1f}s)"
|
||
else:
|
||
line = f" {frame} {self.message} ({elapsed:.1f}s)"
|
||
pad = max(self.last_line_len - len(line), 0)
|
||
self._write(f"\r{line}{' ' * pad}", end='', flush=True)
|
||
self.last_line_len = len(line)
|
||
self.frame_idx += 1
|
||
time.sleep(0.12)
|
||
|
||
def start(self):
|
||
if self.running:
|
||
return
|
||
self.running = True
|
||
self.start_time = time.time()
|
||
self.thread = threading.Thread(target=self._animate, daemon=True)
|
||
self.thread.start()
|
||
|
||
def update_text(self, new_message: str):
|
||
self.message = new_message
|
||
|
||
def _clear_line_blanks(self) -> str:
|
||
# Clear with spaces (not \033[K) to avoid garbled escapes under patch_stdout.
|
||
return ' ' * max(self.last_line_len + 5, 40)
|
||
|
||
def print_above(self, text: str):
|
||
"""Print a line above the spinner; the next tick redraws the spinner below it.
|
||
|
||
Works inside redirect_stdout(devnull) because _write targets the stdout
|
||
captured at spinner creation, not the current sys.stdout.
|
||
"""
|
||
if not self.running:
|
||
self._write(f" {text}", flush=True)
|
||
return
|
||
self._write(f"\r{self._clear_line_blanks()}\r {text}", flush=True)
|
||
|
||
def stop(self, final_message: str = None):
|
||
self.running = False
|
||
if self.thread:
|
||
self.thread.join(timeout=0.5)
|
||
is_tty = self._is_tty
|
||
if is_tty:
|
||
self._write(f"\r{self._clear_line_blanks()}\r", end='', flush=True)
|
||
if final_message:
|
||
elapsed = f" ({time.time() - self.start_time:.1f}s)" if self.start_time else ""
|
||
if is_tty:
|
||
self._write(f" {final_message}", flush=True)
|
||
else:
|
||
self._write(f" [done] {final_message}{elapsed}", flush=True)
|
||
|
||
def __enter__(self):
|
||
self.start()
|
||
return self
|
||
|
||
def __exit__(self, *exc):
|
||
self.stop()
|
||
return False
|
||
|
||
|
||
# =========================================================================
|
||
# Cute tool message (completion line that replaces the spinner)
|
||
# =========================================================================
|
||
|
||
_ERROR_SUFFIX_MAX_LEN = 48
|
||
|
||
|
||
def _trim_error(msg: str) -> str:
|
||
"""Shrink an error message for inline display (long 'File not found' paths -> filename)."""
|
||
msg = msg.strip()
|
||
if "File not found:" in msg:
|
||
tail = msg.partition("File not found:")[2].strip()
|
||
if "/" in tail:
|
||
msg = f"File not found: {tail.rsplit('/', 1)[-1]}"
|
||
return _tail_trunc(msg, _ERROR_SUFFIX_MAX_LEN)
|
||
|
||
|
||
def _detect_tool_failure(tool_name: str, result: str | None) -> tuple[bool, str]:
|
||
"""Return ``(is_failure, suffix)`` for a tool result, e.g. ``(True, " [exit 1]")``."""
|
||
if result is None or file_mutation_result_landed(tool_name, result):
|
||
return False, ""
|
||
data = safe_json_loads(result)
|
||
|
||
# Terminal: non-zero exit code is the canonical failure signal.
|
||
if tool_name == "terminal":
|
||
if isinstance(data, dict):
|
||
exit_code = data.get("exit_code")
|
||
if exit_code is not None and exit_code != 0:
|
||
err_msg = data.get("error")
|
||
return True, f" [{_trim_error(str(err_msg))}]" if err_msg else f" [exit {exit_code}]"
|
||
return False, ""
|
||
|
||
if isinstance(data, dict):
|
||
# Memory: distinguish "store full" from real errors.
|
||
if tool_name == "memory" and data.get("success") is False and "exceed the limit" in data.get("error", ""):
|
||
return True, " [full]"
|
||
err = data.get("error") or data.get("message")
|
||
if err and (data.get("success") is False or "error" in data):
|
||
return True, f" [{_trim_error(str(err))}]"
|
||
|
||
# Multimodal results (dicts) are successes; failures arrive as JSON-encoded strings.
|
||
if not isinstance(result, str):
|
||
return False, ""
|
||
lower = result[:500].lower()
|
||
if '"error"' in lower or '"failed"' in lower or result.startswith("Error"):
|
||
return True, " [error]"
|
||
return False, ""
|
||
|
||
|
||
def _domain(url: str) -> str:
|
||
return url.replace("https://", "").replace("http://", "").split("/")[0]
|
||
|
||
|
||
def _cute_trunc(s) -> str:
|
||
"""Tail-truncate to the configured preview cap (0 = unlimited)."""
|
||
return _tail_trunc(str(s), _tool_preview_max_len)
|
||
|
||
|
||
def _cute_path(p) -> str:
|
||
"""Head-truncate a path to the configured preview cap, keeping the filename end."""
|
||
p = str(p)
|
||
limit = _tool_preview_max_len
|
||
return ("..." + p[-(limit-3):]) if limit and len(p) > limit else p
|
||
|
||
|
||
def _cute_web_extract(a: dict, _r) -> str:
|
||
urls = a.get("urls", [])
|
||
if urls:
|
||
url = _display_url(urls[0] if isinstance(urls, list) else urls)
|
||
if url:
|
||
extra = f" +{len(urls)-1}" if isinstance(urls, list) and len(urls) > 1 else ""
|
||
return f"┊ 📄 fetch {_cute_trunc(_domain(url))}{extra}"
|
||
return "┊ 📄 fetch pages"
|
||
|
||
|
||
def _cute_todo_list(a: dict, result) -> str:
|
||
todos_arg = a.get("todos")
|
||
total = done = 0
|
||
if result:
|
||
try:
|
||
data = safe_json_loads(result)
|
||
if data:
|
||
s = data.get("summary", {})
|
||
total = s.get("total", 0)
|
||
done = s.get("completed", 0)
|
||
except Exception:
|
||
pass
|
||
if todos_arg is None:
|
||
detail = f"{done}/{total} task(s)" if total > 0 else "reading tasks"
|
||
elif a.get("merge", False):
|
||
detail = f"update {done}/{total} ✓" if total > 0 and done > 0 else f"update {len(todos_arg)} task(s)"
|
||
else:
|
||
detail = f"{done}/{total} task(s)" if total > 0 and done > 0 else f"{len(todos_arg)} task(s)"
|
||
return f"┊ 📋 plan {detail}"
|
||
|
||
|
||
def _cute_memory(a: dict, _r) -> str:
|
||
action = a.get("action", "?")
|
||
target = a.get("target", "")
|
||
if action == "add":
|
||
return f"┊ 🧠 memory +{target}: \"{_cute_trunc(a.get('content', ''))}\""
|
||
if action in ("replace", "remove"):
|
||
old = a.get("old_text") or "<missing old_text>"
|
||
return f"┊ 🧠 memory {'~' if action == 'replace' else '-'}{target}: \"{_cute_trunc(old)}\""
|
||
return f"┊ 🧠 memory {action}"
|
||
|
||
|
||
def _cute_skill_view(a: dict, _r) -> str:
|
||
label = a.get("name", "")
|
||
file_path = a.get("file_path")
|
||
if file_path:
|
||
label = f"{label} → {file_path}" if label else str(file_path)
|
||
return f"┊ 📚 skill {_cute_trunc(label)}"
|
||
|
||
|
||
def _cute_cronjob(a: dict, _r) -> str:
|
||
action = a.get("action", "?")
|
||
if action == "create":
|
||
skills = a.get("skills") or ([] if not a.get("skill") else [a.get("skill")])
|
||
label = a.get("name") or (skills[0] if skills else None) or a.get("prompt", "task")
|
||
return f"┊ ⏰ cron create {_cute_trunc(label)}"
|
||
if action == "list":
|
||
return "┊ ⏰ cron listing"
|
||
return f"┊ ⏰ cron {action} {a.get('job_id', '')}"
|
||
|
||
|
||
def _cute_execute_code(a: dict, _r) -> str:
|
||
code = a.get("code", "")
|
||
first_line = code.strip().split("\n")[0] if code.strip() else ""
|
||
return f"┊ 🐍 exec {_cute_trunc(first_line)}"
|
||
|
||
|
||
def _cute_browser_exec(a: dict, _r) -> str:
|
||
# Leading `# …` comment becomes the step label; code stays collapsed behind the preview cap.
|
||
label = _browser_exec_step_label(a)
|
||
if label is not None:
|
||
return f"┊ 🌐 browser {label}"
|
||
return f"┊ 🌐 browser {_cute_trunc(' '.join(str(a.get('code', '') or '').split()))}"
|
||
|
||
|
||
def _cute_delegate(a: dict, _r) -> str:
|
||
action_preview = _delegate_action_preview(a)
|
||
if action_preview is not None:
|
||
return f"┊ 🔀 delegate {_cute_trunc(action_preview)}"
|
||
tasks = a.get("tasks")
|
||
if tasks and isinstance(tasks, list):
|
||
goals = _delegate_task_goals(tasks, per_goal_len=30)
|
||
detail = " | ".join(goals) if goals else "parallel"
|
||
return f"┊ 🔀 delegate {len(goals) or len(tasks)}x: {_cute_trunc(detail)}"
|
||
return f"┊ 🔀 delegate {_cute_trunc(a.get('goal', ''))}"
|
||
|
||
|
||
def _cute_process_manage(a: dict, _r) -> str:
|
||
action = a.get("action", "?")
|
||
sid = a.get("session_id", "")[:12]
|
||
return f"┊ ⚙️ proc {'ls processes' if action == 'list' else f'{action} {sid}'}"
|
||
|
||
|
||
_SCROLL_ARROWS = {"down": "↓", "up": "↑", "right": "→", "left": "←"}
|
||
|
||
# Completion-line renderers: tool -> f(args, result) -> "┊ {emoji} {verb:9} {detail}" (duration appended by caller).
|
||
_CUTE_LINES = {
|
||
"web_search": lambda a, r: f"┊ 🔍 search {_cute_trunc(a.get('query', ''))}",
|
||
"web_extract": _cute_web_extract,
|
||
"terminal": lambda a, r: f"┊ 💻 $ {_cute_trunc(build_tool_preview('terminal', a) or a.get('command', ''))}",
|
||
"process_manage": _cute_process_manage,
|
||
"read_file": lambda a, r: f"┊ 📖 read {_cute_trunc(build_tool_preview('read_file', a) or a.get('path', ''))}",
|
||
"write_file": lambda a, r: f"┊ ✍️ write {_cute_path(a.get('path', ''))}",
|
||
"patch": lambda a, r: f"┊ 🔧 patch {_cute_path(a.get('path', ''))}",
|
||
"search_files": lambda a, r: f"┊ 🔎 {'find' if a.get('target', 'content') == 'files' else 'grep':9} {_cute_trunc(a.get('pattern', ''))}",
|
||
"browser_navigate": lambda a, r: f"┊ 🌐 navigate {_cute_trunc(_domain(a.get('url', '')))}",
|
||
"browser_snapshot": lambda a, r: f"┊ 📸 snapshot {'full' if a.get('full') else 'compact'}",
|
||
"browser_click": lambda a, r: f"┊ 👆 click {a.get('ref', '?')}",
|
||
"browser_type": lambda a, r: f"┊ ⌨️ type \"{_cute_trunc(a.get('text', ''))}\"",
|
||
"browser_scroll": lambda a, r: f"┊ {_SCROLL_ARROWS.get(a.get('direction', 'down'), '↓')} scroll {a.get('direction', 'down')}",
|
||
"browser_back": lambda a, r: "┊ ◀️ back ",
|
||
"browser_press": lambda a, r: f"┊ ⌨️ press {a.get('key', '?')}",
|
||
"browser_get_images": lambda a, r: "┊ 🖼️ images extracting",
|
||
"browser_vision": lambda a, r: "┊ 👁️ vision analyzing page",
|
||
"todo_list": _cute_todo_list,
|
||
"session_search": lambda a, r: f"┊ 🔍 recall \"{_cute_trunc(a.get('query', ''))}\"",
|
||
"memory": _cute_memory,
|
||
"skills_list": lambda a, r: f"┊ 📚 skills list {a.get('category', 'all')}",
|
||
"skill_view": _cute_skill_view,
|
||
"image_generate": lambda a, r: f"┊ 🎨 create {_cute_trunc(a.get('prompt', ''))}",
|
||
"text_to_speech": lambda a, r: f"┊ 🔊 speak {_cute_trunc(a.get('text', ''))}",
|
||
"vision_analyze": lambda a, r: f"┊ 👁️ vision {_cute_trunc(a.get('question', ''))}",
|
||
"send_message": lambda a, r: f"┊ 📨 send {a.get('target', '?')}: \"{_cute_trunc(a.get('message', ''))}\"",
|
||
"cronjob_manage": _cute_cronjob,
|
||
"execute_code": _cute_execute_code,
|
||
"browser_exec": _cute_browser_exec,
|
||
"delegate_task": _cute_delegate,
|
||
}
|
||
|
||
|
||
def _get_cute_tool_message(
|
||
tool_name: str, args: dict, duration: float, result: str | None = None,
|
||
) -> str:
|
||
"""Formatted tool completion line for CLI quiet mode: ``| {emoji} {verb:9} {detail} {duration}``.
|
||
|
||
Failed tool calls get an informational suffix from :func:`_detect_tool_failure`;
|
||
the leading ``┊`` is swapped for the active skin's tool prefix.
|
||
"""
|
||
args = redact_tool_args_for_display(tool_name, args) or args
|
||
is_failure, failure_suffix = _detect_tool_failure(tool_name, result)
|
||
skin_prefix = get_skin_tool_prefix()
|
||
render = _CUTE_LINES.get(tool_name)
|
||
if render is not None:
|
||
body = render(args, result)
|
||
else:
|
||
body = f"┊ ⚡ {tool_name[:9]:9} {_cute_trunc(build_tool_preview(tool_name, args) or '')}"
|
||
line = f"{body} {duration:.1f}s"
|
||
if skin_prefix != "┊":
|
||
line = line.replace("┊", skin_prefix, 1)
|
||
return line if not is_failure else f"{line}{failure_suffix}"
|
||
|
||
|
||
def get_cute_tool_message(
|
||
tool_name: str, args: dict, duration: float, result: str | None = None,
|
||
) -> str:
|
||
"""Render a completion label without letting cosmetic failures escape."""
|
||
try:
|
||
return _get_cute_tool_message(tool_name, args, duration, result=result)
|
||
except Exception as exc: # noqa: BLE001 — display must never abort a turn
|
||
logger.debug("Tool completion label failed for %s: %s", tool_name, exc)
|
||
safe_name = tool_name[:9] if isinstance(tool_name, str) and tool_name else "tool"
|
||
safe_duration = f"{duration:.1f}s" if isinstance(duration, (int, float)) else "done"
|
||
return f"┊ ⚡ {safe_name:9} completed {safe_duration}"
|