hermes-agent/agent/display.py

"""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, replace
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, is_guardrail_refusal

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."""
    value = (value.get("url") or value.get("href")) if isinstance(value, dict) else value
    return value.strip() if isinstance(value, str) else ""


def _http_url(candidate: str) -> str | None:
    """``candidate`` when it parses as an absolute http(s) URL, else None."""
    try:
        parsed = urlsplit(candidate)
    except ValueError:
        return None
    return candidate if parsed.scheme.lower() in {"http", "https"} and parsed.netloc else None


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():
    """Active skin config, or None when unavailable (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


# Foreground diff colors: key -> (skin color key, dark-terminal fallback RGB).
_DIFF_FG = {
    "dim": ("banner_dim", (150, 150, 150)), "file": ("session_label", (180, 160, 255)),
    "hunk": ("session_border", (120, 120, 140)),
}
# Background diff colors (white text on a dark tint of the skin's ui_error/ui_ok):
# key -> (skin key, default hex, dominant channel index, dark-terminal fallback).
_DIFF_BG = {
    "minus": ("ui_error", "#ef5350", 0, "\033[38;2;255;255;255;48;2;120;20;20m"),
    "plus": ("ui_ok", "#4caf50", 1, "\033[38;2;255;255;255;48;2;20;90;20m"),
}


def _fg(r: int, g: int, b: int) -> str:
    return f"\033[38;2;{r};{g};{b}m"


def _tinted_bg(rgb: tuple[int, int, int], dominant: int) -> str:
    r, g, b = (max(v // 2, 20) if i == dominant else max(v // 4, 10) for i, v in enumerate(rgb))
    return f"\033[38;2;255;255;255;48;2;{r};{g};{b}m"


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
    colors = {k: _fg(*rgb) for k, (_, rgb) in _DIFF_FG.items()} | {k: v[3] for k, v in _DIFF_BG.items()}
    try:
        skin = _get_skin()
        for key, (skin_key, fallback) in _DIFF_FG.items():
            h = skin.get_color(skin_key, "")
            colors[key] = _fg(*(_hex_rgb(h) if h and len(h) == 7 and h[0] == "#" else fallback))
        bg_hex = {key: skin.get_color(skin_key, default) for key, (skin_key, default, _, _) in _DIFF_BG.items()}
        for key, (_, _, dominant, _) in _DIFF_BG.items():
            h = bg_hex[key]
            if h and len(h) == 7:
                colors[key] = _tinted_bg(_hex_rgb(h), dominant)
    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
_friendly_tool_labels: bool = True


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 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_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()
    override = skin.tool_emojis.get(tool_name) if skin and skin.tool_emojis else None
    if override:
        return override
    try:
        from tools.registry import registry
        return registry.get_emoji(tool_name, default="") or default
    except Exception:
        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 | None) -> str:
    """Tail-truncate to ``limit`` chars with ``...`` (0/None = unlimited). The result never
    exceeds ``limit``: for 1-3 the ellipsis itself is clipped (``text[:limit - 3]`` would go
    negative and hand back almost the whole string, #9439)."""
    if not limit or limit <= 0 or len(text) <= limit:
        return text
    return "." * limit if limit <= 3 else text[:limit - 3] + "..."


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]:
    parts: list[list[str]] = [[]]
    for _, ch, quoted in _scan_quoted(segment):
        if not quoted and ch.isspace():
            parts.append([])
        else:
            parts[-1].append(ch)
    return ["".join(p) for p in parts if p]


def _strip_shell_pipe_tail(segment: str) -> str:
    """Drop a trailing ``| head/tail/wc/sort/uniq ...`` pipeline stage."""
    words = _split_shell_words(segment)
    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:
            words = words[:i]
            break
    return " ".join(words).strip()


def _split_shell_compound(command: str) -> list[str]:
    """Split on unquoted ``&&`` / ``||`` / ``;`` / newline, dropping pipe tails per segment."""
    raw: list[list[str]] = [[]]
    skip = False
    for i, ch, quoted in _scan_quoted(command):
        if skip:
            skip = False
        elif not quoted and (command.startswith("&&", i) or command.startswith("||", i)):
            raw.append([])
            skip = True
        elif not quoted and ch in {";", "\n"}:
            raw.append([])
        else:
            raw[-1].append(ch)
    segments = (_strip_shell_pipe_tail("".join(buf).strip()) for buf in raw)
    return [s for s in segments if s]


def _shell_head_word(segment: str) -> str:
    """Command name of a segment, skipping leading ``VAR=value`` assignments."""
    words = _split_shell_words(segment)
    while words and re.match(r"^[A-Za-z_]\w*=", words[0]):
        words.pop(0)
    return _shell_basename(words[0] if 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  # operator + target
        elif re.match(r"^\d*(?:>&|<&)\d+$", word):
            i += 1
        else:
            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
    count = len(core) - 1
    return core[0] if not count else f"{core[0]} + {count} {'command' if count == 1 else 'commands'}"


def _read_file_line_label(args: dict) -> str:
    offset, limit = args.get("offset"), args.get("limit")
    if not isinstance(offset, int) or offset <= 0:
        return ""
    return f"L{offset}-{offset + limit - 1}" if isinstance(limit, int) and limit > 1 else f"L{offset}"


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/metadata, so it is swapped before
    reaching logs, callbacks, the model, or chat history. Forced regardless of
    ``security.redact_secrets``: a leaked typed credential is a security boundary.
    """
    needle = "" if typed_text is None else str(typed_text)
    redacted = redact_sensitive_text(needle, force=True) if needle else needle
    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 []
    raw_goals = (task.get("goal") for task in tasks if isinstance(task, dict))
    return [_tail_trunc(("?" if g is None else _oneline(str(g))) or "?", per_goal_len) for g in raw_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."""
    first = str(args.get("code", "") or "").strip().split("\n", 1)[0].strip()
    label = first.lstrip("#").strip() if first.startswith("#") else ""
    if not label:
        return None
    return label[: max_chars - 1] + "…" if len(label) > max_chars else 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 _tail_trunc(label, max_len)
    return _tail_trunc(_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)
    tasks = args.get("tasks")
    if action_preview is not None:
        return _tail_trunc(action_preview, max_len)
    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 _tail_trunc(preview, max_len)
    goal = args.get("goal", "")
    return None if goal is None else _tail_trunc(_oneline(str(goal)), max_len) or None


def _preview_process_manage(args: dict, _max_len: int) -> str | None:
    action, sid, data, timeout_val = (args.get(k) for k in ("action", "session_id", "data", "timeout"))
    parts = [str(action) if action else "", str(sid)[:16] if sid else "",
             f'"{_oneline(str(data)[:20])}"' if data else "", f"{timeout_val}s" if timeout_val and action == "wait" else ""]
    return " ".join(p for p in parts if p) or None


def _preview_todo_list(args: dict, _max_len: int) -> str:
    todos_arg = args.get("todos")
    verb = "updating" if args.get("merge", False) else "planning"
    return "reading task list" if todos_arg is None else f"{verb} {len(todos_arg)} task(s)"


def _preview_shell(key: str):
    def _build(args: dict, max_len: int) -> str | None:
        command = args.get(key)
        return None if command is None else _tail_trunc(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")
    label = (Path(str(path).replace("\\", "/")).name or str(path)) if path is not None else None
    return None if label is None else _tail_trunc(f"{label} {_read_file_line_label(args)}".strip(), max_len) or None


def _preview_memory(args: dict, _max_len: int) -> str:
    action, target = args.get("action", ""), 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:
    return f"to {args.get('target', '?')}: \"{_tail_trunc(_oneline(args.get('message', '')), 20)}\""


def _preview_skill_view(args: dict, max_len: int) -> str | None:
    name = _oneline(str(args.get("name") or ""))
    file_path = args.get("file_path")
    label = (f"{name} → {_oneline(str(file_path))}" if name else _oneline(str(file_path))) if file_path else name
    return _tail_trunc(label, max_len) or None


def _preview_bridge_call(tool_name: str):
    def _build(args: dict, max_len: int) -> str | None:
        labels = bridge_tool_labels(tool_name, args)
        if not labels:
            return _primary_arg_preview(tool_name, args, max_len)
        extra = f" +{len(labels) - 1}" if len(labels) > 1 else ""
        return _tail_trunc(f"{labels[0].text}{extra}", max_len) or None
    return _build


# 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, "memory": _preview_memory, "send_message": _preview_send_message,
    "skill_view": _preview_skill_view,
    "session_search": lambda args, _m: f"recall: \"{_clip(_oneline(args.get('query', '')), 25)}\"",
    "tool_call": _preview_bridge_call("tool_call"),
    "tool_search": _preview_bridge_call("tool_search"),
    "tool_describe": _preview_bridge_call("tool_describe"),
}


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)
    return _primary_arg_preview(tool_name, args, max_len)


def _primary_arg_preview(tool_name: str, args: dict, max_len: int) -> str | None:
    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]
    preview = _oneline(str((value[0] if value else "") if isinstance(value, list) else 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:
    """Compact preview plus explicit truncation/URL facts (the uncapped preview is
    rebuilt from the arguments so an upstream display cap cannot drop its link target)."""
    full_text = build_tool_preview(tool_name, args, max_len=0) or fallback
    text = _tail_trunc(full_text, max_len)
    truncated = text != full_text
    url = _http_url(_display_url(full_text)) if truncated else None
    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"})

_BRIDGE_GENERATING = {
    "tool_call": "a tool call", "tool_search": "a tool search", "tool_describe": "tool details",
}


def bridge_generating_phrase(tool_name: str) -> str | None:
    return _BRIDGE_GENERATING.get(tool_name) if _friendly_tool_labels else None


def tool_labels_for_call(tool_name: str, args: dict | None) -> list:
    try:
        from tools.tool_labels import labels_for_call
        labels = labels_for_call(tool_name, args or {})
    except Exception as exc:  # noqa: BLE001 — display must never abort a turn
        logger.debug("bridge labels failed for %s: %s", tool_name, exc)
        return []
    skin = _get_skin()
    overrides = (skin.tool_emojis if skin and skin.tool_emojis else None) or {}
    return [replace(label, emoji=overrides[label.name]) if label.name in overrides else label
            for label in labels]


def bridge_tool_labels(tool_name: str, args: dict | None) -> list:
    return tool_labels_for_call(tool_name, args) if _friendly_tool_labels else []


def tool_row_emoji(tool_name: str, args: dict | None = None, default: str = "⚡") -> str:
    labels = bridge_tool_labels(tool_name, args) if args else []
    return labels[0].emoji if labels else get_tool_emoji(tool_name, default)


def get_tool_verb(tool_name: str) -> str | None:
    """Friendly verb for a built-in tool, or None (labels disabled / no curated verb);
    callers compose ``f"{verb}{tool_verb_connector(tool)}{preview}"`` themselves."""
    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 following the bot's display name (Slack setStatus).

    ``args=None`` gives a verb-only phrase (``display.live_status: verb`` keeps previews out
    of shared channels). None for ``_thinking`` or disabled labels so callers use 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}"
    with_preview = args and verb and tool_name not in _TOOL_VERBS_NO_PREVIEW
    preview = build_tool_preview(tool_name, args, max_len=None) if with_preview else 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()}"
    return phrase[: max_len - 2].rstrip() + "…" if len(phrase) > max_len - 1 else phrase + "…"


def build_tool_label(tool_name: str, args: dict, max_len: int | None = None) -> str | None:
    labels = bridge_tool_labels(tool_name, args)
    if labels:
        label = labels[0]
        preview = _tail_trunc(label.preview, max_len if max_len is not None else _tool_preview_max_len)
        extra = f" +{len(labels) - 1}" if len(labels) > 1 else ""
        return f"{label.text}{f' {preview}' if preview else ''}{extra}"
    verb = get_tool_verb(tool_name)
    if verb and tool_name in _TOOL_VERBS_NO_PREVIEW:
        return verb
    preview = build_tool_preview(tool_name, args, max_len=max_len)
    if not verb:
        return preview
    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."""
    return Path.cwd() / Path(os.path.expanduser(path))  # a `/` with an absolute rhs keeps the rhs


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 (UnicodeDecodeError, OSError):  # FileNotFoundError/IsADirectoryError are OSErrors
        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 == "delete":
        return [path for path in sorted(skill_dir.rglob("*")) if path.is_file()]
    if file_path and action in {"edit", "patch", "write_file", "remove_file"}:
        return [skill_dir / file_path]
    return [skill_dir / "SKILL.md"] if action in {"edit", "patch"} else []


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 == "skill_manage":
        return _resolve_skill_manage_paths(function_args)
    path = function_args.get("path") if tool_name in {"write_file", "patch"} else None
    return [_resolved_path(path)] if path else []


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."""
    data = safe_json_loads(result) if result else None
    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, after = snapshot.before.get(str(path)), _snapshot_text(path)
        if before == after:
            continue
        display_path = _display_diff_path(path)
        diff = "".join(unified_diff(
            (before or "").splitlines(keepends=True), (after or "").splitlines(keepends=True),
            fromfile=f"a/{display_path}", tofile=f"b/{display_path}",
        ))
        if diff:
            chunks.append(diff if diff.endswith("\n") else diff + "\n")
    return "".join(chunks) or None


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)
        diff = data.get("diff") if isinstance(data, dict) else None
        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:
        for line in ["  ┊ review diff", *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 = 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]] = [[]]
    for line in diff.splitlines():
        if line.startswith("--- ") and sections[-1]:
            sections.append([])
        sections[-1].append(line)
    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 = 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': list('⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏'), 'bounce': list('⠁⠂⠄⡀⢀⠠⠐⠈'), 'grow': list('▁▂▃▄▅▆▇█▇▆▅▄▃▂'),
        'arrows': list('←↖↑↗→↘↓↙'), 'star': list('✶✷✸✹✺✹✸✷'), 'moon': list('🌑🌒🌓🌔🌕🌖🌗🌘'),
        'pulse': list('◜◠◝◞◡◟'), 'brain': list('🧠💭💡✨💫🌟💡💭'), 'sparkle': list('⁺˚*✧✦✧*˚'),
    }
    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()
            return (skin.spinner.get(key, []) if skin else None) or fallback
        except Exception:
            return fallback

    get_waiting_faces = classmethod(lambda cls: cls._skin_spinner_list("waiting_faces", cls.KAWAII_WAITING))
    get_thinking_faces = classmethod(lambda cls: cls._skin_spinner_list("thinking_faces", cls.KAWAII_THINKING))
    get_thinking_verbs = classmethod(lambda cls: 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 = self.start_time = None
        self.frame_idx = self.last_line_len = 0
        self._print_fn = print_fn  # when set, bypasses self._out so silenced agents stay silent
        self._out = sys.stdout  # captured NOW, before any child redirect_stdout(devnull) replaces it

    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 for prompt_toolkit's StdoutProxy: it injects newlines around each flush so the
        \\r overwrite never lands, and the CLI drives its own TUI spinner widget in that mode."""
        try:
            from prompt_toolkit.patch_stdout import StdoutProxy
            return isinstance(self._out, StdoutProxy)
        except ImportError:
            return False

    def _animate(self):
        tty = self._is_tty
        # Non-TTY (Docker, systemd, pipe): log once instead of spamming frames.
        if not tty:
            self._write(f"  [tool] {self.message}", flush=True)
        # Under patch_stdout the \r animation would overdraw the TUI status bar.
        if not tty or self._is_patch_stdout_proxy():
            while self.running:
                time.sleep(0.5 if not tty else 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 = f"({time.time() - self.start_time:.1f}s)"
            left, right = wings[self.frame_idx % len(wings)] if wings else ("", "")
            line = f"  {left} {frame} {self.message} {right} {elapsed}" if wings else f"  {frame} {self.message} {elapsed}"
            self._write(f"\r{line}{' ' * max(self.last_line_len - len(line), 0)}", 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, self.start_time = True, 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:
        return ' ' * max(self.last_line_len + 5, 40)  # spaces, not \033[K: garbles under patch_stdout

    def print_above(self, text: str):
        """Print a line above the spinner (next tick redraws it below). Works inside
        redirect_stdout(devnull) because _write targets the stdout captured at creation."""
        self._write(f"\r{self._clear_line_blanks()}\r  {text}" if self.running else f"  {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 ""
            self._write(f"  {final_message}" if is_tty else 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
# A degraded backend (Docker down, SSH host unreachable) needs the whole reason plus the fix hint.
_DEGRADED_SUFFIX_MAX_LEN = 200


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 _degraded_suffix(data: dict) -> str:
    """`` [<reason> — <retry_hint>]`` for a ``status: degraded`` terminal result (hint omitted when empty)."""
    reason = str(data.get("reason") or data.get("error") or "terminal backend unavailable").strip()
    hint = str(data.get("retry_hint") or "").strip()
    text = f"{reason} — {hint}" if hint else reason
    return f" [{_tail_trunc(text, _DEGRADED_SUFFIX_MAX_LEN)}]"


def _detect_tool_failure(tool_name: str, result: Any) -> 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 = result if isinstance(result, dict) else safe_json_loads(result)
    # A harness REFUSAL of a redundant call (repeated identical read/search) is not a
    # failed call. This is the ``failed`` the executor hands the loop guardrail, so
    # counting it would escalate refusals into ``repeated_exact_failure_block``.
    if is_guardrail_refusal(data):
        return False, ""

    # A denied/timed-out approval carries one human sentence; show it instead of the model-facing
    # "BLOCKED: ... Do NOT retry" text (which stays in the JSON for the model).
    if isinstance(data, dict) and data.get("user_summary"):
        return True, f" [{_tail_trunc(str(data['user_summary']), _DEGRADED_SUFFIX_MAX_LEN)}]"

    # Terminal: non-zero exit code is the canonical failure signal.
    if tool_name == "terminal":
        exit_code = data.get("exit_code") if isinstance(data, dict) else None
        if exit_code is None or exit_code == 0:
            return False, ""
        if data.get("status") == "degraded":
            return True, _degraded_suffix(data)
        err_msg = data.get("error")
        return True, f" [{_trim_error(str(err_msg))}]" if err_msg else f" [exit {exit_code}]"

    if isinstance(data, dict):
        failed = data.get("success") is False
        # Memory: distinguish "store full" from real errors.
        if tool_name == "memory" and failed and "exceed the limit" in data.get("error", ""):
            return True, " [full]"
        err = data.get("error") or data.get("message")
        if err and (failed or "error" in data):
            return True, f" [{_trim_error(str(err))}]"
    # Multimodal results (dicts) are successes; failures arrive as JSON-encoded strings.
    if isinstance(result, str) and (
        '"error"' in result[:500].lower() or '"failed"' in result[:500].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
    if not limit or len(p) <= limit:
        return p
    return "." * limit if limit <= 3 else "..." + p[-(limit - 3):]


def _cute_web_extract(a: dict, _r) -> str:
    urls = a.get("urls", [])
    url = _display_url(urls[0] if isinstance(urls, list) else urls) if urls else ""
    if not url:
        return "┊ 📄 fetch     pages"
    extra = f" +{len(urls)-1}" if isinstance(urls, list) and len(urls) > 1 else ""
    return f"┊ 📄 fetch     {_cute_trunc(_domain(url))}{extra}"


def _cute_todo_list(a: dict, result) -> str:
    todos_arg = a.get("todos")
    total = done = 0
    try:
        summary = (safe_json_loads(result) or {}).get("summary", {}) if result else {}
        total, done = summary.get("total", 0), summary.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, target = a.get("action", "?"), 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, file_path = a.get("name", ""), a.get("file_path")
    label = (f"{label} → {file_path}" if label else str(file_path)) if file_path else label
    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 ([a.get("skill")] if a.get("skill") else [])
        label = a.get("name") or (skills[0] if skills else None) or a.get("prompt", "task")
        return f"┊ ⏰ cron      create {_cute_trunc(label)}"
    return "┊ ⏰ cron      listing" if action == "list" else f"┊ ⏰ cron      {action} {a.get('job_id', '')}"


def _cute_execute_code(a: dict, _r) -> str:
    code = a.get("code", "").strip()
    return f"┊ 🐍 exec      {_cute_trunc(code.split(chr(10))[0] if code else '')}"


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)
    return f"┊ 🌐 browser   {_cute_trunc(_oneline(str(a.get('code', '') or ''))) if label is None else label}"


def _cute_delegate(a: dict, _r) -> str:
    action_preview = _delegate_action_preview(a)
    tasks = a.get("tasks")
    if action_preview is not None:
        return f"┊ 🔀 delegate  {_cute_trunc(action_preview)}"
    if tasks and isinstance(tasks, list):
        goals = _delegate_task_goals(tasks, per_goal_len=30)
        return f"┊ 🔀 delegate  {len(goals) or len(tasks)}x: {_cute_trunc(' | '.join(goals) if goals else 'parallel')}"
    return f"┊ 🔀 delegate  {_cute_trunc(a.get('goal', ''))}"


def _cute_process_manage(a: dict, _r) -> str:
    action, sid = a.get("action", "?"), 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 _cute_bridge_rows(tool_name: str, args: dict) -> list[str] | None:
    labels = bridge_tool_labels(tool_name, args)
    if not labels:
        return None
    return [f"┊ {label.emoji} {_cute_trunc(label.text)}"
            f"{f'  {_cute_trunc(label.preview)}' if label.preview else ''}" for label in labels]


_BRIDGE_CALL_INDEX_RE = re.compile(r"calls\[(\d+)\]")


def _entry_failure_suffix(entry: Any) -> str:
    if not isinstance(entry, dict):
        return ""
    error = entry.get("error")
    if isinstance(error, dict):
        return f" [{_trim_error(str(error.get('message') or error.get('code') or 'error'))}]"
    if not (error or entry.get("success") is False):
        return ""
    return _detect_tool_failure(str(entry.get("name") or ""), entry)[1]


def _bridge_row_suffixes(rows: int, result: Any, call_suffix: str) -> list[str]:
    suffixes = [""] * rows
    data = result if isinstance(result, dict) else safe_json_loads(result)
    entries = data.get("results") if isinstance(data, dict) else None
    if isinstance(entries, list):
        for position, entry in enumerate(entries[:rows]):
            suffix = _entry_failure_suffix(entry)
            if not suffix:
                continue
            index = entry.get("index")
            suffixes[index if isinstance(index, int) and 0 <= index < rows else position] = suffix
    if call_suffix and not any(suffixes):
        match = _BRIDGE_CALL_INDEX_RE.search(call_suffix)
        named = int(match.group(1)) if match else rows - 1
        suffixes[named if 0 <= named < rows else rows - 1] = call_suffix
    return suffixes


def _get_cute_tool_message(tool_name: str, args: dict, duration: float, result: str | None = None) -> str:
    args = redact_tool_args_for_display(tool_name, args) or args
    is_failure, failure_suffix = _detect_tool_failure(tool_name, result)
    render = _CUTE_LINES.get(tool_name)
    rows = _cute_bridge_rows(tool_name, args)
    if rows is None:
        body = render(args, result) if render else f"┊ ⚡ {tool_name[:9]:9} {_cute_trunc(build_tool_preview(tool_name, args) or '')}"
        rows, suffixes = [body], [failure_suffix if is_failure else ""]
    else:
        suffixes = _bridge_row_suffixes(len(rows), result, failure_suffix if is_failure else "")
    rows = [*rows[:-1], f"{rows[-1]}  {duration:.1f}s"]
    prefix = get_skin_tool_prefix()
    return "\n  ".join(f"{row}{suffix}".replace("┊", prefix, 1) for row, suffix in zip(rows, suffixes))


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}"


# ---- BEGIN PLUGIN-COMPAT (revert-scheduled; see COMPAT_MANIFEST.md) ----
# Names external plugins imported from this module before the Sep 2026 decomposition.
# Internal code MUST NOT use these (scripts/check_compat_pointers.py fails CI if it does).
# The whole block is removed by reverting the commit that added it.

def get_friendly_tool_labels() -> bool:
    """Return whether friendly tool labels are enabled."""
    return _friendly_tool_labels
# ---- END PLUGIN-COMPAT ----