diff --git a/agent/display.py b/agent/display.py index 94259fa80a899..e5f80226b6acf 100644 --- a/agent/display.py +++ b/agent/display.py @@ -29,6 +29,28 @@ _MAX_INLINE_DIFF_FILES = 6 _MAX_INLINE_DIFF_LINES = 80 +# Set to True by the CLI when code-highlight mode is active. Consumed by +# get_cute_tool_message to suppress the inline code snippet (the highlighted +# block will show the full code immediately after). +_code_highlight_active: bool = False + + +def set_code_highlight_active(active: bool) -> None: + global _code_highlight_active + _code_highlight_active = active + + +# Rich-based rendering (syntax highlighting + enhanced diffs) +try: + from agent.rich_output import DiffRenderer as _RichDiffRenderer + from agent.rich_output import SyntaxHighlighter as _RichSyntaxHighlighter + from agent.rich_output import clean_command_output + _rich_diff = _RichDiffRenderer() + _rich_syntax = _RichSyntaxHighlighter() + _RICH_OUTPUT = True +except ImportError: + _RICH_OUTPUT = False + @dataclass class LocalEditSnapshot: @@ -411,7 +433,18 @@ def _emit_inline_diff(diff_text: str, print_fn) -> bool: def _render_inline_unified_diff(diff: str) -> list[str]: - """Render unified diff lines in Hermes' inline transcript style.""" + """Render unified diff lines with line numbers and coloured backgrounds. + + Uses rich_output.DiffRenderer when available (line numbers, green/red + background highlights). Falls back to the original ANSI-string path. + """ + if _RICH_OUTPUT: + try: + return _rich_diff.to_lines(diff) + except Exception as exc: + logger.debug("Rich diff render failed, using ANSI fallback: %s", exc) + + # Original ANSI fallback — unchanged from upstream rendered: list[str] = [] from_file = None to_file = None @@ -443,6 +476,24 @@ def _render_inline_unified_diff(diff: str) -> list[str]: return rendered +def highlight_code( + code: str, + language: str | None = None, + filename: str | None = None, +) -> str: + """Return an ANSI-highlighted version of *code* for terminal display. + + When rich_output is unavailable the original string is returned unchanged. + """ + if not _RICH_OUTPUT: + return code + try: + return _rich_syntax.to_ansi(code, language=language, filename=filename) + except Exception as exc: + logger.debug("highlight_code failed: %s", exc) + return code + + def _split_unified_diff_sections(diff: str) -> list[str]: """Split a unified diff into per-file sections.""" sections: list[list[str]] = [] @@ -531,6 +582,193 @@ def render_edit_diff_with_delta( return _emit_inline_diff("\n".join(rendered_lines), print_fn) +# ========================================================================= +# execute_code / read_file / terminal syntax highlight previews +# ========================================================================= + +_HIGHLIGHT_MAX_LINES = 40 + + +def _emit_highlighted_lines(lines: list[str], print_fn) -> None: + """Print *lines*, truncating at _HIGHLIGHT_MAX_LINES with a dim footer.""" + _print = print_fn or print + if len(lines) <= _HIGHLIGHT_MAX_LINES: + for line in lines: + _print(line) + else: + for line in lines[:_HIGHLIGHT_MAX_LINES]: + _print(line) + omitted = len(lines) - _HIGHLIGHT_MAX_LINES + _print(f"\033[2m ╌╌ {omitted} more line{'s' if omitted != 1 else ''} omitted ╌╌\033[0m") + + +def _highlight_block(header: str, content: str, language: str, print_fn) -> bool: + """Print a labelled syntax-highlighted block aligned with the ┊ tool log. + + Format:: + + \033[2m ┊
\033[0m + + """ + _print = print_fn or print + _print(f"\033[2m ┊ {header}\033[0m") + if not _RICH_OUTPUT: + _emit_highlighted_lines(content.rstrip("\n").splitlines(), _print) + return True + try: + highlighted = _rich_syntax.to_ansi(content, language=language).rstrip("\n") + _emit_highlighted_lines(highlighted.splitlines(), _print) + return True + except Exception as exc: + logger.debug("highlight_block failed for %s: %s", header, exc) + return False + + +def render_execute_code_preview(code: str, print_fn=None) -> bool: + """Print *code* with Python syntax highlighting. + + The cute_msg line already labels the tool; this function prints only the + highlighted code (no header) so the output stays compact. + Returns True if anything was printed. + """ + if not code or not code.strip(): + return False + _print = print_fn or print + if not _RICH_OUTPUT: + _emit_highlighted_lines(code.rstrip("\n").splitlines(), _print) + return True + try: + highlighted = _rich_syntax.to_ansi(code, language="python").rstrip("\n") + _emit_highlighted_lines(highlighted.splitlines(), _print) + return True + except Exception as exc: + logger.debug("execute_code highlight failed: %s", exc) + return False + + +def render_read_file_preview(path: str, result_json: str, print_fn=None) -> bool: + """Print the content of a read_file result with syntax highlighting. + + Language is detected from *path*'s extension. Returns False (no output) + when the file type is unknown — we don't highlight plain text or binary. + """ + if not path or not result_json: + return False + try: + import json as _json + result = _json.loads(result_json) + content = result.get("content", "") + except Exception: + return False + if not content or not content.strip(): + return False + + from pathlib import Path as _Path + if _RICH_OUTPUT: + from agent.rich_output import LanguageDetector as _LD + lang = _LD().detect_from_filename(_Path(path).name) + else: + lang = None + if not lang: + return False # unknown type — skip, don't guess + + header = f"📄 {_Path(path).name}" + return _highlight_block(header, content, lang, print_fn) + + +_FILE_READ_COMMANDS = frozenset({ + "cat", "head", "tail", "less", "more", "bat", + "sed", "awk", "grep", "cut", "sort", "uniq", + "nl", "od", "xxd", "hexdump", +}) + +# Commands that *execute* a file rather than reading it — the terminal output +# will be runtime stdout, not source code. Never highlight for these. +_FILE_EXEC_COMMANDS = frozenset({ + "python", "python3", "python2", + "node", "nodejs", "deno", "bun", + "ruby", "perl", "php", "lua", + "bash", "sh", "zsh", "fish", "dash", + "Rscript", "julia", +}) + + +def _extract_file_language_from_command(command: str): + """Return (filename, language) if *command* is clearly reading a known source file. + + Only fires when the leading verb is a known file-reader (cat, head, sed …). + Commands that *execute* files (node, python, bash …) are explicitly excluded + — their stdout is runtime output, not source code. + + Parses tokens in reverse (file arg is typically last) and returns the first + token whose extension maps to a known language. Returns (None, None) if no + match — we never fall back to content-based detection for shell output. + """ + if not command: + return None, None + try: + import shlex as _shlex + tokens = _shlex.split(command) + except ValueError: + tokens = command.split() + + if not tokens: + return None, None + + # Check the leading verb (strip path prefix, e.g. /usr/bin/cat → cat) + from pathlib import Path as _Path + verb = _Path(tokens[0]).name + if verb in _FILE_EXEC_COMMANDS: + return None, None + if _FILE_READ_COMMANDS and verb not in _FILE_READ_COMMANDS: + # Not in the explicit read list — only proceed if it's clearly not an + # executor. Unknown commands might be aliases like 'bat' equivalents; + # allow them through so we don't over-block. + pass + + if _RICH_OUTPUT: + from agent.rich_output import LanguageDetector as _LD + detector = _LD() + else: + return None, None + + for tok in reversed(tokens): + if tok.startswith("-"): + continue + # Only consider tokens that look like a file path (contain a dot or slash) + if "." not in _Path(tok).name: + continue + lang = detector.detect_from_filename(_Path(tok).name) + if lang: + return _Path(tok).name, lang + return None, None + + +def render_terminal_preview(command: str, result_json: str, print_fn=None) -> bool: + """Print terminal output with syntax highlighting when the command reads a source file. + + Highlighting is only applied when a known-extension filename can be extracted + from *command* (e.g. ``cat foo.py``, ``sed -n '1,50p' app.ts``). + Returns False without printing anything if the language cannot be determined. + """ + if not command or not result_json: + return False + filename, lang = _extract_file_language_from_command(command) + if not lang: + return False + try: + import json as _json + result = _json.loads(result_json) + output = result.get("output", "") + except Exception: + return False + if not output or not output.strip(): + return False + + header = f"💻 {filename}" + return _highlight_block(header, output, lang, print_fn) + + # ========================================================================= # KawaiiSpinner # ========================================================================= @@ -950,6 +1188,8 @@ def _wrap(line: str) -> str: } return _wrap(f"┊ 🧪 rl {rl.get(tool_name, tool_name.replace('rl_', ''))} {dur}") if tool_name == "execute_code": + if _code_highlight_active: + return _wrap(f"┊ 🐍 exec {dur}") code = args.get("code", "") first_line = code.strip().split("\n")[0] if code.strip() else "" return _wrap(f"┊ 🐍 exec {_trunc(first_line, 35)} {dur}") diff --git a/agent/rich_output.py b/agent/rich_output.py new file mode 100644 index 0000000000000..a83d667685440 --- /dev/null +++ b/agent/rich_output.py @@ -0,0 +1,974 @@ +"""Rich-based syntax highlighting, diff rendering, and output utilities. + +Drop into hermes-agent's ``agent/`` directory. + +No project-specific imports — only ``rich`` (always present in Hermes) and +``pygments`` (bundled as a rich dependency). + +Public API +---------- +LanguageDetector detect language from filename / content +FilePathFormatter per-type icons + compact relative-path display +SyntaxHighlighter Pygments → Rich markup → ANSI string +DiffRenderer unified diff → Rich Text with line numbers → ANSI lines +apply_inline_markdown convert **bold** / *italic* / `code` / ~~strike~~ to ANSI +apply_block_line convert block-level markdown (headings, hr, blockquotes, + lists) to ANSI on a single line +clean_command_output strip venv/stacktrace noise from command output +""" + +from __future__ import annotations + +import logging +import os +import re +import shutil +from difflib import SequenceMatcher +from io import StringIO +from pathlib import Path +from typing import Optional + +from rich.console import Console, Group +from rich.style import Style +from rich.text import Text + +logger = logging.getLogger(__name__) + +# Diff background colours — kept in sync with agent.display._ANSI_PLUS / _ANSI_MINUS +# _ANSI_PLUS = "\033[38;2;255;255;255;48;2;20;90;20m" → rgb(20,90,20) +# _ANSI_MINUS = "\033[38;2;255;255;255;48;2;120;20;20m" → rgb(120,20,20) +_DIFF_BG_ADD = "#145a14" # rgb(20, 90, 20) +_DIFF_BG_DEL = "#781414" # rgb(120, 20, 20) + +# Minimum SequenceMatcher ratio to apply intra-line highlighting. +# Below this the lines are too dissimilar and highlighting would be noise. +_INTRA_DIFF_MIN_RATIO: float = 0.5 + +# --------------------------------------------------------------------------- +# Pygments availability (bundled transitively via rich, but guard anyway) +# --------------------------------------------------------------------------- + +try: + from pygments.lexers import ( + TextLexer, + get_lexer_by_name, + get_lexer_for_filename, + guess_lexer, + ) + from pygments.token import ( + Comment, + Error, + Generic, + Keyword, + Name, + Number, + Operator, + String, + ) + from pygments.util import ClassNotFound + + _PYGMENTS = True +except ImportError: + _PYGMENTS = False + + +# --------------------------------------------------------------------------- +# Language detection +# --------------------------------------------------------------------------- + +class LanguageDetector: + """Detect programming language from filename extension or code content.""" + + EXTENSION_MAP: dict[str, str] = { + # Python + ".py": "python", ".pyx": "python", ".pyi": "python", ".pyw": "python", + # JavaScript / TypeScript + ".js": "javascript", ".jsx": "jsx", ".mjs": "javascript", ".cjs": "javascript", + ".ts": "typescript", ".tsx": "tsx", + # JVM + ".java": "java", ".scala": "scala", ".kt": "kotlin", ".groovy": "groovy", + # C family + ".c": "c", ".h": "c", + ".cpp": "cpp", ".cxx": "cpp", ".cc": "cpp", ".hpp": "cpp", + ".cs": "csharp", ".fs": "fsharp", + # Systems + ".rs": "rust", ".go": "go", ".swift": "swift", + # Scripting + ".rb": "ruby", ".php": "php", ".pl": "perl", ".lua": "lua", + ".r": "r", ".R": "r", + # Web + ".html": "html", ".htm": "html", ".css": "css", ".scss": "scss", + ".sass": "sass", ".vue": "vue", ".svelte": "svelte", + # Shell + ".sh": "bash", ".bash": "bash", ".zsh": "zsh", ".fish": "fish", + ".ps1": "powershell", ".bat": "batch", ".cmd": "batch", + # Data / config + ".json": "json", ".yaml": "yaml", ".yml": "yaml", + ".toml": "toml", ".ini": "ini", ".cfg": "ini", ".xml": "xml", + # Docs + ".md": "markdown", ".rst": "rst", ".tex": "latex", + # DB + ".sql": "sql", + # Containers + ".dockerfile": "dockerfile", + # Other + ".dart": "dart", ".ex": "elixir", ".exs": "elixir", + ".erl": "erlang", ".hs": "haskell", ".ml": "ocaml", + ".elm": "elm", ".zig": "zig", ".vim": "vim", + } + + CONTENT_PATTERNS: dict[str, list[str]] = { + "python": [ + r"^\s*def\s+\w+\s*\(", r"^\s*class\s+\w+\s*[\(:]", + r"^\s*import\s+\w+", r"^\s*from\s+\w+\s+import", + r'if\s+__name__\s*==\s*[\'"]__main__[\'"]', + ], + "javascript": [ + r"^\s*function\s+\w+\s*\(", r"^\s*const\s+\w+\s*=", + r"console\.log\s*\(", r'require\s*\([\'"]', r"module\.exports", + ], + "typescript": [ + r"^\s*interface\s+\w+", r"^\s*type\s+\w+\s*=", + r":\s*string\s*[;,}]", r":\s*number\s*[;,}]", + ], + "java": [r"^\s*public\s+class\s+\w+", r"System\.out\.print"], + "cpp": [r"#include\s*<\w+>", r"std::\w+", r"cout\s*<<"], + "go": [r"^\s*package\s+\w+", r"^\s*func\s+\w+\s*\(", r"fmt\.Print"], + "rust": [r"^\s*fn\s+\w+\s*\(", r"^\s*use\s+\w+", r"println!\s*\("], + "bash": [r"#!/bin/bash", r"#!/bin/sh", r"^\s*if\s*\[", r"\$\{\w+\}"], + "sql": [r"^\s*SELECT\s+", r"^\s*INSERT\s+INTO", r"^\s*CREATE\s+TABLE"], + } + + def detect_from_filename(self, filename: str) -> Optional[str]: + if not filename: + return None + ext = Path(filename).suffix.lower() + if ext in self.EXTENSION_MAP: + return self.EXTENSION_MAP[ext] + name = Path(filename).name.lower() + if name in {"dockerfile", "makefile", "rakefile", "gemfile", "vagrantfile"}: + return name + return None + + def detect_from_content(self, content: str, max_lines: int = 50) -> Optional[str]: + if not content.strip(): + return None + sample = "\n".join(content.split("\n")[:max_lines]) + scores: dict[str, int] = {} + for lang, patterns in self.CONTENT_PATTERNS.items(): + score = sum(1 for p in patterns if re.search(p, sample, re.MULTILINE)) + if score: + scores[lang] = score + return max(scores, key=lambda k: scores[k]) if scores else None + + def detect(self, content: str, filename: Optional[str] = None) -> Optional[str]: + return self.detect_from_filename(filename) or self.detect_from_content(content) + + +# --------------------------------------------------------------------------- +# File path formatting +# --------------------------------------------------------------------------- + +class FilePathFormatter: + """Per-filetype icons and compact relative-path display.""" + + _ICONS: dict[str, str] = { + ".py": "🐍", ".js": "📜", ".ts": "📘", ".tsx": "⚛️", ".jsx": "⚛️", + ".html": "🌐", ".css": "🎨", ".scss": "🎨", ".md": "📝", + ".json": "📋", ".yaml": "⚙️", ".yml": "⚙️", ".toml": "⚙️", + ".txt": "📄", ".log": "📊", ".conf": "⚙️", ".cfg": "⚙️", + ".xml": "📋", ".sql": "🗃️", ".sh": "💻", ".bash": "💻", + ".go": "🐹", ".rs": "🦀", ".java": "☕", ".cpp": "⚙️", + ".c": "⚙️", ".h": "📋", + } + + @staticmethod + def get_file_icon(file_path: str) -> str: + ext = os.path.splitext(file_path)[1].lower() + return FilePathFormatter._ICONS.get(ext, "📄") + + @staticmethod + def format_path( + file_path: str, + compact: bool = True, + cwd: Optional[str] = None, + ) -> str: + if not compact: + return file_path + try: + return os.path.relpath(file_path, cwd or os.getcwd()) + except (ValueError, OSError): + return file_path + + @staticmethod + def titled( + file_path: str, + compact: bool = True, + cwd: Optional[str] = None, + ) -> str: + """Return ``{icon} {path}`` string.""" + icon = FilePathFormatter.get_file_icon(file_path) + path = FilePathFormatter.format_path(file_path, compact, cwd) + return f"{icon} {path}" + + +# --------------------------------------------------------------------------- +# Pygments → Rich markup formatter (internal) +# --------------------------------------------------------------------------- + +class _PygmentsToRich: + """Convert a Pygments token stream to a Rich markup string.""" + + # Built lazily so the class-level dict isn't populated when Pygments is absent + _STYLES: dict = {} + + @classmethod + def _ensure_styles(cls) -> None: + if cls._STYLES or not _PYGMENTS: + return + cls._STYLES = { + Keyword: "bold blue", + Keyword.Type: "bold cyan", + Name: "white", + Name.Builtin: "cyan", + Name.Class: "bold yellow", + Name.Constant: "bold yellow", + Name.Decorator: "bright_cyan", + Name.Exception: "bold red", + Name.Function: "bold yellow", + Name.Function.Magic: "cyan", + Name.Tag: "bold blue", + Name.Variable.Magic: "cyan", + Comment: "dim green", + Comment.Preproc: "bold green", + String: "green", + String.Doc: "dim green", + String.Escape: "bold green", + String.Interpol: "bold green", + String.Regex: "magenta", + Number: "magenta", + Operator: "white", + Operator.Word: "bold blue", + Generic.Deleted: "red", + Generic.Inserted: "green", + Generic.Error: "bold red", + Error: "bold red", + } + + def format(self, tokens) -> str: + self._ensure_styles() + parts: list[str] = [] + for ttype, value in tokens: + style = self._resolve(ttype) + if style and value.strip(): + esc = value.replace("[", r"\[").replace("]", r"\]") + parts.append(f"[{style}]{esc}[/{style}]") + else: + parts.append(value) + return "".join(parts) + + def _resolve(self, ttype) -> Optional[str]: + t = ttype + while t is not None: + if t in self._STYLES: + return self._STYLES[t] + t = t.parent # type: ignore[assignment] + return None + + +# --------------------------------------------------------------------------- +# Public: syntax highlighter +# --------------------------------------------------------------------------- + +class SyntaxHighlighter: + """Highlight source code using Pygments, output as Rich markup or ANSI. + + Falls back to plain green when Pygments is unavailable. + """ + + def __init__(self) -> None: + self._fmt = _PygmentsToRich() + self._detector = LanguageDetector() + + # -- Rich markup (for embedding in Rich Text / Panel) -------------------- + + def to_markup( + self, + code: str, + language: Optional[str] = None, + filename: Optional[str] = None, + ) -> str: + """Return a Rich markup string with syntax colours applied.""" + if not _PYGMENTS: + escaped = code.replace("[", r"\[").replace("]", r"\]") + return f"[green]{escaped}[/green]" + try: + lexer = self._lexer(code, language, filename) + return self._fmt.format(list(lexer.get_tokens(code))) + except Exception as exc: + logger.debug("Pygments highlight failed: %s", exc) + escaped = code.replace("[", r"\[").replace("]", r"\]") + return f"[green]{escaped}[/green]" + + # -- ANSI string (for plain print / print_fn) ---------------------------- + + def to_ansi( + self, + code: str, + language: Optional[str] = None, + filename: Optional[str] = None, + ) -> str: + """Return an ANSI-escaped string suitable for plain ``print()``.""" + markup = self.to_markup(code, language, filename) + buf = StringIO() + Console(file=buf, highlight=False, force_terminal=True, width=220).print(markup) + return buf.getvalue() + + # -- Helpers ------------------------------------------------------------- + + def _lexer(self, code: str, language: Optional[str], filename: Optional[str]): + try: + if language: + return get_lexer_by_name(language, stripnl=False) + if filename: + return get_lexer_for_filename(filename, stripnl=False) + return guess_lexer(code, stripnl=False) + except ClassNotFound: + return TextLexer(stripnl=False) + + +# --------------------------------------------------------------------------- +# Diff renderer helpers (module-level for testability) +# --------------------------------------------------------------------------- + +def _parse_diff_filename(path: str, fallback: Optional[str] = None) -> str: + """Return the basename from a unified-diff path string. + + Strips ``b/`` / ``a/`` prefixes produced by ``git diff``. If the result + is ``/dev/null`` (deleted-file diff), recurses on *fallback* (the ``---`` + path) instead. + """ + for prefix in ("b/", "a/"): + if path.startswith(prefix): + path = path[len(prefix):] + break + if path == "/dev/null": + if fallback: + return _parse_diff_filename(fallback) + return "?" + name = Path(path).name + return name if name else path + + +def _count_pass( + lines: list[str], + explicit_filename: Optional[str] = None, +) -> list[tuple[Optional[str], int, int]]: + """First pass over diff lines: build ``(filename, n_adds, n_dels)`` per file boundary. + + When *explicit_filename* is provided (from ``from_content()``), the + filename is fixed and only one entry is produced. Otherwise filenames are + parsed from ``+++ `` lines. + """ + entries: list[tuple[Optional[str], int, int]] = [] + current_file: Optional[str] = explicit_filename + n_adds = n_dels = 0 + from_path: Optional[str] = None + started = explicit_filename is not None + + for line in lines: + if line.startswith("--- "): + from_path = line[4:].strip() + elif line.startswith("+++ "): + if started: + entries.append((current_file, n_adds, n_dels)) + to_path = line[4:].strip() + if explicit_filename is None: + current_file = _parse_diff_filename(to_path, from_path) + n_adds = n_dels = 0 + started = True + elif line.startswith("+"): + n_adds += 1 + elif line.startswith("-"): + n_dels += 1 + + if started: + entries.append((current_file, n_adds, n_dels)) + + return entries + + +def _make_header(filename: Optional[str], n_adds: int, n_dels: int) -> tuple[Text, Text]: + """Return ``(header_Text, separator_Text)`` for the diff summary line.""" + def _pl(n: int) -> str: + return f"{n} line" if n == 1 else f"{n} lines" + + parts: list[Text] = [ + Text("● ", style="bright_white"), + Text(filename or "?", style=Style(color="bright_white", bold=True)), + Text(" "), + ] + if n_adds > 0 and n_dels == 0: + parts.append(Text(f"Added {_pl(n_adds)}", style="green")) + elif n_dels > 0 and n_adds == 0: + parts.append(Text(f"Removed {_pl(n_dels)}", style="red")) + elif n_adds > 0 and n_dels > 0: + parts.append(Text(f"Added {_pl(n_adds)}", style="green")) + parts.append(Text(f", removed {_pl(n_dels)}", style="red")) + + header = Text.assemble(*parts) + separator = Text("─" * len(header.plain), style="dim") + return header, separator + + +def _flat_del(ln: int, content: str) -> Text: + """Render a deletion line with flat (no intra-line) highlighting.""" + return Text.assemble( + Text(f"{ln:>4} ", style="dim"), + Text("- ", style=Style(color="red", bold=True)), + Text(content, style=Style(bgcolor=_DIFF_BG_DEL, color="white")), + ) + + +def _flat_add(ln: int, content: str) -> Text: + """Render an addition line with flat (no intra-line) highlighting.""" + return Text.assemble( + Text(f"{ln:>4} ", style="dim"), + Text("+ ", style=Style(color="green", bold=True)), + Text(content, style=Style(bgcolor=_DIFF_BG_ADD, color="white")), + ) + + +def _intra_diff(old: str, new: str) -> tuple[list[Text], list[Text]]: + """Character-level diff between two line content strings. + + Returns ``(del_segments, add_segments)`` — lists of ``Text`` objects + covering the full content of each line with no gaps. Changed characters + are rendered bright-red / bright-green bold; unchanged characters use the + base diff background with white foreground. + + Callers: ``Text.assemble(*del_segments)`` / ``Text.assemble(*add_segments)``. + + Note: segment lists may have different total character counts when + ``delete`` or ``insert`` opcodes are present — this is correct because the + two lines have different lengths. + """ + del_segs: list[Text] = [] + add_segs: list[Text] = [] + for tag, i1, i2, j1, j2 in SequenceMatcher(None, old, new, autojunk=False).get_opcodes(): + if tag == "equal": + del_segs.append(Text(old[i1:i2], style=Style(bgcolor=_DIFF_BG_DEL, color="white"))) + add_segs.append(Text(new[j1:j2], style=Style(bgcolor=_DIFF_BG_ADD, color="white"))) + elif tag == "replace": + del_segs.append(Text(old[i1:i2], style=Style(bgcolor=_DIFF_BG_DEL, color="bright_red", bold=True))) + add_segs.append(Text(new[j1:j2], style=Style(bgcolor=_DIFF_BG_ADD, color="bright_green", bold=True))) + elif tag == "delete": + del_segs.append(Text(old[i1:i2], style=Style(bgcolor=_DIFF_BG_DEL, color="bright_red", bold=True))) + elif tag == "insert": + add_segs.append(Text(new[j1:j2], style=Style(bgcolor=_DIFF_BG_ADD, color="bright_green", bold=True))) + return del_segs, add_segs + + +# --------------------------------------------------------------------------- +# Public: diff renderer +# --------------------------------------------------------------------------- + +class DiffRenderer: + """Render a unified diff as Rich Text objects with line numbers. + + Produces coloured ``+`` / ``-`` lines with green / red backgrounds and + dim context lines — significantly richer than raw ANSI strings. + """ + + # -- From old/new strings ------------------------------------------------ + + def from_content( + self, + old: str, + new: str, + file_path: str = "file", + context_lines: int = 3, + ) -> Group: + """Generate and render a diff between *old* and *new*.""" + import difflib + + lines = list(difflib.unified_diff( + old.splitlines(keepends=False), + new.splitlines(keepends=False), + fromfile=f"a/{file_path}", + tofile=f"b/{file_path}", + n=context_lines, + lineterm="", + )) + return self._style(lines, file_path=file_path) + + # -- From unified diff text ---------------------------------------------- + + def from_unified(self, diff_text: str) -> Group: + """Render an already-generated unified diff string.""" + return self._style(diff_text.splitlines()) + + # -- ANSI lines (drop-in for _render_inline_unified_diff) ---------------- + + def to_lines(self, diff_text: str, width: int = 220) -> list[str]: + """Render *diff_text* and return a list of ANSI-escaped strings. + + Compatible with Hermes's ``print_fn`` pattern: each element maps to + one ``print_fn(line)`` call. + """ + buf = StringIO() + Console(file=buf, highlight=False, force_terminal=True, width=width).print( + self.from_unified(diff_text) + ) + # Drop the trailing empty line that Console adds + return buf.getvalue().rstrip("\n").splitlines() + + # -- Internal rendering -------------------------------------------------- + + def _style(self, lines: list[str], file_path: Optional[str] = None) -> Group: + """Render *lines* (from a unified diff) as a ``Group`` of Rich ``Text``. + + *file_path* — when supplied (from ``from_content()``), its basename is + used for the summary header instead of parsing the ``+++ `` line. + """ + styled: list[Text] = [] + + # Pass 1 — count adds/dels per file boundary for the summary header. + explicit_filename = Path(file_path).name if file_path else None + file_entries = iter(_count_pass(lines, explicit_filename)) + + # Pass 2 — render with run-based pairing and intra-line highlighting. + ln_old = ln_new = 0 + from_path: Optional[str] = None + del_run: list[tuple[int, str]] = [] # (line_number, content) + add_run: list[tuple[int, str]] = [] + + def flush_runs() -> None: + """Pair del/add runs and emit highlighted (or flat) Text objects.""" + if not del_run and not add_run: + return + n_pairs = min(len(del_run), len(add_run)) + + # Precompute intra-diff segments for each pair. + pair_segs: list[tuple[Optional[list[Text]], Optional[list[Text]]]] = [] + for i in range(n_pairs): + old_content = del_run[i][1] + new_content = add_run[i][1] + r = SequenceMatcher(None, old_content, new_content).ratio() + if r >= _INTRA_DIFF_MIN_RATIO: + d, a = _intra_diff(old_content, new_content) + pair_segs.append((d, a)) + else: + pair_segs.append((None, None)) + + for i, (ln, content) in enumerate(del_run): + if i < n_pairs and pair_segs[i][0] is not None: + styled.append(Text.assemble( + Text(f"{ln:>4} ", style="dim"), + Text("- ", style=Style(color="red", bold=True)), + *pair_segs[i][0], + )) + else: + styled.append(_flat_del(ln, content)) + + for i, (ln, content) in enumerate(add_run): + if i < n_pairs and pair_segs[i][1] is not None: + styled.append(Text.assemble( + Text(f"{ln:>4} ", style="dim"), + Text("+ ", style=Style(color="green", bold=True)), + *pair_segs[i][1], + )) + else: + styled.append(_flat_add(ln, content)) + + del_run.clear() + add_run.clear() + + for line in lines: + if line.startswith("--- "): + flush_runs() + from_path = line[4:].strip() + continue + + if line.startswith("+++ "): + flush_runs() + entry = next(file_entries, None) + if entry: + fname, n_adds, n_dels = entry + header, sep = _make_header(fname, n_adds, n_dels) + styled.append(header) + styled.append(sep) + continue + + if line.startswith("@@"): + flush_runs() + m = re.search(r"@@ -(\d+),?\d* \+(\d+),?\d* @@", line) + if m: + ln_old, ln_new = int(m.group(1)), int(m.group(2)) + styled.append(Text(line, style=Style(color="cyan", bold=True))) + continue + + if line.startswith("-"): + if add_run: + # -→+→- transition: flush current run and start fresh + flush_runs() + del_run.append((ln_old, line[1:])) + ln_old += 1 + continue + + if line.startswith("+"): + add_run.append((ln_new, line[1:])) + ln_new += 1 + continue + + # Context line — show new-file line number (matches GitHub/delta convention + # and avoids duplicate numbers when old/new offsets diverge) + flush_runs() + content = line[1:] if line.startswith(" ") else line + styled.append(Text.assemble( + Text(f"{ln_new:>4} ", style="dim"), + Text(" ", style="dim"), + Text(content, style="dim"), + )) + ln_old += 1 + ln_new += 1 + + flush_runs() # end of input + styled.append(Text("")) # trailing blank line + return Group(*styled) + + +# --------------------------------------------------------------------------- +# Public: inline markdown → ANSI rendering +# --------------------------------------------------------------------------- + +_MD_CODE_RE = re.compile(r"`([^`\n]+)`") +_MD_BOLD_STAR_RE = re.compile(r"\*\*(.+?)\*\*") +_MD_BOLD_UNDER_RE = re.compile(r"(?(.*?)", re.IGNORECASE) +_MD_STRONG_RE = re.compile(r"(.*?)", re.IGNORECASE) + +_MD_BOLD_ANSI = "\033[1m" +_MD_ITALIC_ANSI = "\033[3m" +_MD_STRIKE_ANSI = "\033[9m" +_MD_CODE_ANSI = "\033[97m" +_MD_RST_ANSI = "\033[0m" + + +def apply_inline_markdown(line: str, reset_suffix: str = "") -> str: + """Apply ANSI styling to inline markdown spans in a single text line. + + Handles ``**bold**``, ``__bold__``, ``*italic*``, ``_italic_``, + ``~~strikethrough~~``, and `` `code` ``. Backtick spans are processed + first and their content is protected from bold/italic passes via + placeholder tokens. + + ``reset_suffix`` is appended after each closing reset; pass the active + response-text ANSI colour here so it is restored between adjacent spans + during streaming. + + Returns *line* unchanged if it already contains ANSI escape codes. + """ + if "\x1b" in line: + return line + + rst = _MD_RST_ANSI + reset_suffix + + # Step 1: protect backtick code spans with index placeholders so later + # passes cannot match * or _ inside them. + protected: list[str] = [] + + def _protect_code(m: re.Match) -> str: # type: ignore[type-arg] + protected.append(f"{_MD_CODE_ANSI}{m.group(1)}{rst}") + return f"\x00{len(protected) - 1}\x00" + + line = _MD_CODE_RE.sub(_protect_code, line) + + # Step 2: bold + line = _MD_BOLD_STAR_RE.sub(lambda m: f"{_MD_BOLD_ANSI}{m.group(1)}{rst}", line) + line = _MD_BOLD_UNDER_RE.sub(lambda m: f"{_MD_BOLD_ANSI}{m.group(1)}{rst}", line) + + # Step 3: italic (runs after bold so ** is already consumed) + line = _MD_ITALIC_STAR_RE.sub(lambda m: f"{_MD_ITALIC_ANSI}{m.group(1)}{rst}", line) + line = _MD_ITALIC_UNDER_RE.sub(lambda m: f"{_MD_ITALIC_ANSI}{m.group(1)}{rst}", line) + + # Step 4: strikethrough + line = _MD_STRIKE_RE.sub(lambda m: f"{_MD_STRIKE_ANSI}{m.group(1)}{rst}", line) + + # Step 5a: images (before links — ![ prefix overlaps) + line = _MD_IMAGE_RE.sub(lambda m: f"\033[2m[img: {m.group(1)}]\033[0m{reset_suffix}", line) + + # Step 5b: links — underline text, discard URL + line = _MD_LINK_RE.sub(lambda m: f"\033[4m{m.group(1)} ({m.group(2)})\033[0m{reset_suffix}", line) + + # Step 5c: HTML inline tags + line = _MD_EM_RE.sub(lambda m: f"{_MD_ITALIC_ANSI}{m.group(1)}\033[0m{reset_suffix}", line) + line = _MD_STRONG_RE.sub(lambda m: f"{_MD_BOLD_ANSI}{m.group(1)}\033[0m{reset_suffix}", line) + + # Step 6: restore protected code spans + for idx, span in enumerate(protected): + line = line.replace(f"\x00{idx}\x00", span) + + # Step 7: strip CommonMark backslash escapes (\] → ], \* → *, etc.) + line = re.sub(r'\\([\\`*_{}\[\]()#+\-.!|~])', r'\1', line) + + return line + + +# --------------------------------------------------------------------------- +# Public: block-level markdown → ANSI rendering +# --------------------------------------------------------------------------- + +_MD_HEADING_RE = re.compile(r"^(#{1,6})\s+(.*)") +_MD_HR_RE = re.compile(r"^(-{3,}|\*{3,}|_{3,})$") +_MD_BLOCKQUOTE_RE = re.compile(r"^>+\s?(.*)") +_MD_UL_RE = re.compile(r"^(\s*)([-*+])\s+(.+)") +_MD_REF_LINK_RE = re.compile(r"^\[[^\]]+\]:\s+\S+") + +_HEADING_STYLES = { + 1: "\033[1;97m", + 2: "\033[1;37m", + 3: "\033[1m", + 4: "\033[1;2m", + 5: "\033[1;2m", + 6: "\033[1;2m", +} +_BLOCKQUOTE_ANSI = "\033[2m" +_BULLETS = ["•", "◦", "▸", "·"] + + +def apply_block_line(line: str) -> str: + """Apply ANSI styling to block-level markdown structures in a single line. + + Handles headings (h1–h6), horizontal rules, blockquotes, unordered lists, + and reference link suppression. Ordered lists are passed through unchanged. + + Two early-exit guards: + - Lines containing ``\\x1b`` are already ANSI-rendered — returned as-is. + - Lines containing ``\\n`` are multi-line blocks from ``StreamingBlockBuffer`` + (table or setext) — returned as-is. + + Returns *line* unchanged if no block pattern matches. + """ + if "\x1b" in line: + return line + if "\n" in line: + return line + + # Reference link definition — suppress entirely + if _MD_REF_LINK_RE.match(line): + return "" + + # Headings + m = _MD_HEADING_RE.match(line) + if m: + level = len(m.group(1)) + text = m.group(2) + style = _HEADING_STYLES.get(level, "\033[1;2m") + rendered_text = apply_inline_markdown(text, reset_suffix=style) + return f"{style}{rendered_text}{_MD_RST_ANSI}" + + # Horizontal rule + stripped = line.rstrip() + if _MD_HR_RE.match(stripped): + cols = shutil.get_terminal_size((80, 24)).columns + return f"\033[2m{'─' * cols}\033[0m" + + # Blockquote — collapse any level of nesting to single gutter + m = _MD_BLOCKQUOTE_RE.match(line) + if m: + content = m.group(1) + content_rendered = apply_inline_markdown(content, reset_suffix=_BLOCKQUOTE_ANSI) + return f"{_BLOCKQUOTE_ANSI}▌ {content_rendered}\033[0m" + + # Unordered list — bullet symbol by indent depth + m = _MD_UL_RE.match(line) + if m: + indent, _marker, content = m.group(1), m.group(2), m.group(3) + level = len(indent) // 2 + bullet = _BULLETS[min(level, len(_BULLETS) - 1)] + return f"{indent}{bullet} {content}" + + return line + + +# --------------------------------------------------------------------------- +# Public: fenced code block highlighting for LLM responses +# --------------------------------------------------------------------------- + +def format_response(text: str) -> str: + """Apply syntax highlighting and markdown rendering to a complete response string. + + Pass 1: replaces each `` ```lang\\ncode\\n``` `` block with an + ANSI-highlighted version. Pass 2: applies block-level then inline markdown + (headings, hr, blockquotes, lists, bold, italic, code spans, etc.) to every + non-code line. Suitable for the non-streaming Rich Panel display path. + """ + _hl = SyntaxHighlighter() + _det = LanguageDetector() + + _RST = "\033[0m" # reset — transparent no-op; marks lines as code for pass 2 + + def _highlight(m: "re.Match") -> str: + lang = m.group(2).strip() or None + code = m.group(3) + if not lang: + lang = _det.detect_from_content(code) + highlighted = _hl.to_ansi(code, language=lang).rstrip("\n") + # Some lexers (e.g. plain-text) emit lines with no ANSI codes. + # Pass 2 uses `"\x1b" in l` to detect already-highlighted lines and + # skip markdown rendering. Guarantee every code-block line has at + # least one escape by prepending a no-op reset to bare lines. + lines_out = [] + for line in highlighted.splitlines(): + lines_out.append(line if "\x1b" in line else _RST + line) + return "\n".join(lines_out) + + # Match fenced code blocks of any depth (3+ backticks); \1 backreference + # ensures the closing fence uses the same backtick sequence as the opener. + text = re.sub(r"(?m)^(`{3,})(\w*)\n(.*?)\1", _highlight, text, flags=re.DOTALL) + # Block + inline markdown pass — lines with \x1b are already highlighted code. + # Use splitlines() (no keepends) so apply_block_line never receives a trailing + # \n that its capture groups would silently drop. Rejoin manually and restore + # the final newline if the original text ended with one. + lines = text.splitlines() + result = "\n".join( + l if "\x1b" in l else apply_inline_markdown(apply_block_line(l)) + for l in lines + ) + if text.endswith("\n"): + result += "\n" + return result + + +class StreamingCodeBlockHighlighter: + """State machine that syntax-highlights fenced code blocks during streaming. + + Feed lines one at a time with ``process_line()``. Regular lines are + returned immediately; lines inside a code block are buffered and the + entire highlighted block is returned when the closing fence arrives. + + Example usage in a line-emission loop:: + + hl = StreamingCodeBlockHighlighter() + for line in stream_lines: + out = hl.process_line(line) + if out is not None: + emit(out) + # End of stream — flush any unclosed block + tail = hl.flush() + if tail is not None: + emit(tail) + """ + + # Matches an opening fence: 3+ backticks, optional language hint (word chars) + _FENCE_OPEN_RE = re.compile(r"^(`{3,})\s*(\w*)$") + # Matches a closing fence: 3+ backticks, optional trailing whitespace only + _FENCE_CLOSE_RE = re.compile(r"^(`+)\s*$") + + def __init__(self) -> None: + self._in_block: bool = False + self._lang: Optional[str] = None + self._fence_depth: int = 3 # backtick count of the opening fence + self._buf: list[str] = [] + self._hl = SyntaxHighlighter() + self._det = LanguageDetector() + + def process_line(self, line: str) -> Optional[str]: + """Process one line. + + Returns the string to emit (may be multi-line for a highlighted block), + or ``None`` to suppress the line (still accumulating a code block). + """ + stripped = line.strip() + + if not self._in_block: + m = self._FENCE_OPEN_RE.match(stripped) + if m: + self._in_block = True + self._fence_depth = len(m.group(1)) + self._lang = m.group(2) or None + self._buf = [] + return None # suppress opening fence — will re-emit with block + return line # plain text, pass through + + # Inside a code block — closing fence: >= fence_depth backticks, nothing else + m = self._FENCE_CLOSE_RE.match(stripped) + if m and len(m.group(1)) >= self._fence_depth: + return self._flush_block() + self._buf.append(line) + return None # still accumulating + + def flush(self) -> Optional[str]: + """Flush any open (unclosed) code block at end of stream.""" + if self._in_block and self._buf: + return self._flush_block() + return None + + def reset(self) -> None: + """Reset state for a new response turn.""" + self._in_block = False + self._lang = None + self._fence_depth = 3 + self._buf = [] + + def _flush_block(self) -> str: + code = "\n".join(self._buf) + lang = self._lang or self._det.detect_from_content(code) + highlighted = self._hl.to_ansi(code, language=lang).rstrip("\n") + self._in_block = False + self._lang = None + self._buf = [] + return highlighted + + +# --------------------------------------------------------------------------- +# Public: output noise cleaning +# --------------------------------------------------------------------------- + +_NOISE_SUBSTRINGS = frozenset({ + "/venv/lib/python", "/site-packages/", "langsmith/", "langchain/", + "__pycache__", "venv/lib/", "site-packages", + "Traceback (most recent call last)", ' File "/', +}) + + +def clean_command_output(content: str) -> str: + """Strip venv paths, stacktrace boilerplate, and excessive blank lines. + + Useful for cleaning up ``terminal`` tool results before display. + """ + out: list[str] = [] + for line in content.split("\n"): + line = line.strip() + if not line: + continue + if any(s in line for s in _NOISE_SUBSTRINGS): + continue + if len(line) > 80 and line.count("/") > 5: + continue + if line.startswith("from ") and "import" in line and len(line) > 60: + continue + line = re.sub(r"\\n\./([^/]+/)*", "", line) + line = re.sub(r"\\n/[^/]+/[^/]+/([^/]+)", r" \1", line) + line = line.replace("\\n", "\n").replace("\n\n\n", "\n\n") + if line and len(line) > 3: + out.append(line) + + result = "\n".join(out) + return re.sub(r"\n\s*\n\s*\n", "\n\n", result).strip() + + +# --------------------------------------------------------------------------- +# Module-level convenience singletons +# --------------------------------------------------------------------------- + +lang_detector = LanguageDetector() +syntax_highlighter = SyntaxHighlighter() +diff_renderer = DiffRenderer() diff --git a/cli.py b/cli.py index b13317fe95c09..4c17414ef4d68 100644 --- a/cli.py +++ b/cli.py @@ -460,6 +460,17 @@ def load_cli_config() -> Dict[str, Any]: except Exception: pass +# Rich-based response highlighting (syntax highlight fenced code blocks) +try: + import agent.display as _display + from agent.rich_output import StreamingCodeBlockHighlighter as _CodeBlockHL + from agent.rich_output import apply_block_line as _apply_block_line + from agent.rich_output import apply_inline_markdown as _apply_inline_md + from agent.rich_output import format_response as _format_response + _RICH_RESPONSE = True +except ImportError: + _RICH_RESPONSE = False + # Neuter AsyncHttpxClientWrapper.__del__ before any AsyncOpenAI clients are # created. The SDK's __del__ schedules aclose() on asyncio.get_running_loop() # which, during CLI idle time, finds prompt_toolkit's event loop and tries to @@ -1080,6 +1091,11 @@ def __init__( # Inline diff previews for write actions (display.inline_diffs in config.yaml) self._inline_diffs_enabled = CLI_CONFIG["display"].get("inline_diffs", True) + # Syntax-highlighted code preview for execute_code (display.code_highlight in config.yaml) + self._code_highlight_enabled = CLI_CONFIG["display"].get("code_highlight", True) + from agent.display import set_code_highlight_active + set_code_highlight_active(self._code_highlight_enabled) + # Streaming display state self._stream_buf = "" # Partial line buffer for line-buffered rendering self._stream_started = False # True once first delta arrives @@ -1864,7 +1880,22 @@ def _emit_stream_text(self, text: str) -> None: _tc = getattr(self, "_stream_text_ansi", "") while "\n" in self._stream_buf: line, self._stream_buf = self._stream_buf.split("\n", 1) - _cprint(f"{_tc}{line}{_RST}" if _tc else line) + if _RICH_RESPONSE: + out = self._stream_code_hl.process_line(line) + if out is None: + continue # buffering a code block + if out is line: + # Plain text — apply block + inline markdown then response text colour. + # Always apply when _RICH_RESPONSE is True; _code_highlight_active only + # gates tool-output syntax highlighting, not LLM response rendering. + line = _apply_inline_md(_apply_block_line(line), reset_suffix=_tc) + _cprint(f"{_tc}{line}{_RST}" if _tc else line) + else: + # Highlighted code block — emit as-is (carries its own ANSI) + for hl_line in out.splitlines(): + _cprint(hl_line) + else: + _cprint(f"{_tc}{line}{_RST}" if _tc else line) def _flush_stream(self) -> None: """Emit any remaining partial line from the stream buffer and close the box.""" @@ -1873,7 +1904,18 @@ def _flush_stream(self) -> None: if self._stream_buf: _tc = getattr(self, "_stream_text_ansi", "") - _cprint(f"{_tc}{self._stream_buf}{_RST}" if _tc else self._stream_buf) + if _RICH_RESPONSE: + out = self._stream_code_hl.process_line(self._stream_buf) + if out is not None: + if out is self._stream_buf: + out = _apply_inline_md(_apply_block_line(out), reset_suffix=_tc) + _cprint(f"{_tc}{out}{_RST}" if _tc else out) + # Flush any open code block (unclosed fence at end of response) + tail = self._stream_code_hl.flush() + if tail: + _cprint(tail) + else: + _cprint(f"{_tc}{self._stream_buf}{_RST}" if _tc else self._stream_buf) self._stream_buf = "" # Close the response box @@ -1893,6 +1935,11 @@ def _reset_stream_state(self) -> None: self._reasoning_box_opened = False self._reasoning_buf = "" self._reasoning_preview_buf = "" + if _RICH_RESPONSE: + if not hasattr(self, "_stream_code_hl"): + self._stream_code_hl = _CodeBlockHL() + else: + self._stream_code_hl.reset() def _slow_command_status(self, command: str) -> str: """Return a user-facing status message for slower slash commands.""" @@ -3900,6 +3947,8 @@ def process_command(self, command: str) -> bool: self.console.print(f" Status bar {state}") elif canonical == "verbose": self._toggle_verbose() + elif canonical in ("code-highlight", "codehighlight", "code_highlight"): + self._toggle_code_highlight() elif canonical == "yolo": self._toggle_yolo() elif canonical == "reasoning": @@ -4617,6 +4666,17 @@ def _toggle_verbose(self): } _cprint(labels.get(self.tool_progress_mode, "")) + def _toggle_code_highlight(self): + """Toggle syntax-highlighted code preview for execute_code.""" + self._code_highlight_enabled = not self._code_highlight_enabled + from agent.display import set_code_highlight_active + set_code_highlight_active(self._code_highlight_enabled) + from hermes_cli.colors import Colors as _Colors + if self._code_highlight_enabled: + _cprint(f"{_Colors.GREEN}Code highlight: ON{_Colors.RESET} — execute_code will show syntax-highlighted Python.") + else: + _cprint(f"{_Colors.DIM}Code highlight: OFF{_Colors.RESET} — execute_code preview disabled.") + def _toggle_yolo(self): """Toggle YOLO mode — skip all dangerous command approval prompts.""" import os @@ -5052,8 +5112,16 @@ def _on_tool_start(self, tool_call_id: str, function_name: str, function_args: d logger.debug("Edit snapshot capture failed for %s", function_name, exc_info=True) def _on_tool_complete(self, tool_call_id: str, function_name: str, function_args: dict, function_result: str): - """Render file edits with inline diff after write-capable tools complete.""" + """Render file edits with inline diff / code preview after tools complete. + + Both features are suppressed when tool_progress_mode is "off" — that + mode promises "silent, just the final response". + """ snapshot = self._pending_edit_snapshots.pop(tool_call_id, None) + + if self.tool_progress_mode == "off": + return + try: from agent.display import render_edit_diff_with_delta @@ -5067,6 +5135,24 @@ def _on_tool_complete(self, tool_call_id: str, function_name: str, function_args except Exception: logger.debug("Edit diff preview failed for %s", function_name, exc_info=True) + if self._code_highlight_enabled: + try: + from agent.display import ( + _result_succeeded, + render_execute_code_preview, + render_read_file_preview, + render_terminal_preview, + ) + if function_name == "execute_code": + if _result_succeeded(function_result): + render_execute_code_preview(function_args.get("code", ""), print_fn=_cprint) + elif function_name == "read_file": + render_read_file_preview(function_args.get("path", ""), function_result, print_fn=_cprint) + elif function_name == "terminal": + render_terminal_preview(function_args.get("command", ""), function_result, print_fn=_cprint) + except Exception: + logger.debug("%s highlight failed", function_name, exc_info=True) + # ==================================================================== # Voice mode methods # ==================================================================== @@ -6105,8 +6191,11 @@ def run_agent(): pass else: _chat_console = ChatConsole() + _rendered_response = ( + _format_response(response) if _RICH_RESPONSE else response + ) _chat_console.print(Panel( - _rich_text_from_ansi(response), + _rich_text_from_ansi(_rendered_response), title=f"[{_resp_color} bold]{label}[/]", title_align="left", border_style=_resp_color, diff --git a/tests/test_display.py b/tests/test_display.py index 5127a930ba115..6991e64d400d4 100644 --- a/tests/test_display.py +++ b/tests/test_display.py @@ -1,16 +1,26 @@ """Tests for agent/display.py — build_tool_preview() and inline diff previews.""" +import json import os +import re import pytest from unittest.mock import MagicMock, patch from agent.display import ( + _HIGHLIGHT_MAX_LINES, + _highlight_block, + _result_succeeded, build_tool_preview, capture_local_edit_snapshot, extract_edit_diff, + get_cute_tool_message, _render_inline_unified_diff, _summarize_rendered_diff_sections, render_edit_diff_with_delta, + render_execute_code_preview, + render_read_file_preview, + render_terminal_preview, + set_code_highlight_active, ) @@ -111,10 +121,13 @@ def test_render_inline_unified_diff_colors_added_and_removed_lines(self): " context\n" ) - assert "a/cli.py" in rendered[0] - assert "b/cli.py" in rendered[0] - assert any("old line" in line for line in rendered) - assert any("new line" in line for line in rendered) + import re as _re + _strip = lambda s: _re.sub(r"\x1b\[[0-9;]*m", "", s) + stripped = [_strip(l) for l in rendered] + # v2 header shows basename + change summary, not a/x → b/x + assert "cli.py" in stripped[0] + assert any("old line" in l for l in stripped) + assert any("new line" in l for l in stripped) assert any("48;2;" in line for line in rendered) def test_extract_edit_diff_ignores_non_edit_tools(self): @@ -153,7 +166,8 @@ def test_render_edit_diff_with_delta_invokes_printer(self): assert rendered is True assert printer.call_count >= 2 calls = [call.args[0] for call in printer.call_args_list] - assert any("a/x" in line and "b/x" in line for line in calls) + # v2 header shows basename + change summary + assert any("x" in line for line in calls) assert any("old" in line for line in calls) assert any("new" in line for line in calls) @@ -195,8 +209,278 @@ def test_summarize_rendered_diff_sections_limits_file_count(self): rendered = _summarize_rendered_diff_sections(diff, max_files=3, max_lines=50) - assert any("a/file0.py" in line for line in rendered) - assert any("a/file1.py" in line for line in rendered) - assert any("a/file2.py" in line for line in rendered) - assert not any("a/file7.py" in line for line in rendered) + # v2 header shows basename only + assert any("file0.py" in line for line in rendered) + assert any("file1.py" in line for line in rendered) + assert any("file2.py" in line for line in rendered) + assert not any("file7.py" in line for line in rendered) assert "additional file" in rendered[-1] + + +# --------------------------------------------------------------------------- +# _highlight_block +# --------------------------------------------------------------------------- + +class TestHighlightBlock: + def _collect(self, header, content, language="python"): + calls = [] + _highlight_block(header, content, language, calls.append) + return calls + + def test_header_uses_pipe_prefix(self): + calls = self._collect("📄 foo.py", "x = 1") + assert calls[0].startswith("\033[2m ┊ ") + + def test_header_contains_label(self): + calls = self._collect("📄 foo.py", "x = 1") + assert "📄 foo.py" in calls[0] + + def test_no_separator_line(self): + calls = self._collect("📄 foo.py", "x = 1\ny = 2") + stripped = [re.sub(r"\x1b\[[0-9;]*m", "", c).strip() for c in calls] + # No line should consist solely of box-drawing dashes + assert not any(c and all(ch == "─" for ch in c) for c in stripped) + + def test_returns_true_on_success(self): + assert _highlight_block("hdr", "code", "python", MagicMock()) is True + + def test_returns_false_on_empty_content_exception(self): + # Simulate to_ansi raising — should return False + with patch("agent.display._rich_syntax") as mock_hl, \ + patch("agent.display._RICH_OUTPUT", True): + mock_hl.to_ansi.side_effect = RuntimeError("boom") + result = _highlight_block("hdr", "code", "python", MagicMock()) + assert result is False + + def test_fallback_no_rich_prints_raw_code(self): + calls = [] + with patch("agent.display._RICH_OUTPUT", False): + _highlight_block("hdr", "line one\nline two", "python", calls.append) + assert any("line one" in c for c in calls) + assert any("line two" in c for c in calls) + + +# --------------------------------------------------------------------------- +# render_execute_code_preview +# --------------------------------------------------------------------------- + +class TestRenderExecuteCodePreview: + def test_returns_true_and_emits_code(self): + calls = [] + result = render_execute_code_preview("x = 42", print_fn=calls.append) + assert result is True + assert any("x" in c for c in calls) + + def test_empty_string_returns_false(self): + calls = [] + assert render_execute_code_preview("", print_fn=calls.append) is False + assert calls == [] + + def test_whitespace_only_returns_false(self): + assert render_execute_code_preview(" \n\t ", print_fn=MagicMock()) is False + + def test_no_header_emitted(self): + """Cute-msg already labels the tool — no header should appear in output.""" + calls = [] + render_execute_code_preview("x = 1", print_fn=calls.append) + assert not any("execute_code" in c for c in calls) + + def test_fallback_no_rich_prints_raw_lines(self): + calls = [] + with patch("agent.display._RICH_OUTPUT", False): + result = render_execute_code_preview("a = 1\nb = 2", print_fn=calls.append) + assert result is True + assert any("a = 1" in c for c in calls) + assert any("b = 2" in c for c in calls) + + def test_fallback_no_header_or_separator(self): + calls = [] + with patch("agent.display._RICH_OUTPUT", False): + render_execute_code_preview("a = 1", print_fn=calls.append) + stripped = [re.sub(r"\x1b\[[0-9;]*m", "", c).strip() for c in calls] + assert not any("execute_code" in c for c in stripped) + assert not any(c and all(ch == "─" for ch in c) for c in stripped) + + +# --------------------------------------------------------------------------- +# render_read_file_preview +# --------------------------------------------------------------------------- + +class TestRenderReadFilePreview: + def _result(self, content): + return json.dumps({"content": content}) + + def test_known_extension_returns_true(self): + calls = [] + result = render_read_file_preview("foo.py", self._result("x = 1"), print_fn=calls.append) + assert result is True + assert len(calls) >= 1 + + def test_header_contains_pipe_and_filename(self): + calls = [] + render_read_file_preview("app.ts", self._result("const x = 1;"), print_fn=calls.append) + assert any("┊" in c and "app.ts" in c for c in calls) + + def test_unknown_extension_returns_false(self): + calls = [] + result = render_read_file_preview("notes.log", self._result("some log"), print_fn=calls.append) + assert result is False + assert calls == [] + + def test_empty_content_returns_false(self): + assert render_read_file_preview("foo.py", self._result(""), print_fn=MagicMock()) is False + assert render_read_file_preview("foo.py", self._result(" \n "), print_fn=MagicMock()) is False + + def test_empty_path_returns_false(self): + assert render_read_file_preview("", self._result("x = 1"), print_fn=MagicMock()) is False + + def test_invalid_json_returns_false(self): + assert render_read_file_preview("foo.py", "not json", print_fn=MagicMock()) is False + + +# --------------------------------------------------------------------------- +# render_terminal_preview +# --------------------------------------------------------------------------- + +class TestRenderTerminalPreview: + def _result(self, output): + return json.dumps({"output": output}) + + def test_cat_py_returns_true_and_has_header(self): + calls = [] + result = render_terminal_preview("cat app.py", self._result("x = 1\n"), print_fn=calls.append) + assert result is True + assert any("┊" in c and "app.py" in c for c in calls) + + def test_no_extension_match_returns_false(self): + assert render_terminal_preview("ls -la", self._result("file.txt"), print_fn=MagicMock()) is False + + def test_flag_tokens_skipped_sed(self): + calls = [] + result = render_terminal_preview( + "sed -n '1,50p' app.ts", self._result("const x = 1;"), print_fn=calls.append + ) + assert result is True + assert any("app.ts" in c for c in calls) + + def test_exec_command_node_suppressed(self): + """node script.js executes the file — stdout is not source code.""" + assert render_terminal_preview( + "node script.js", self._result("output text"), print_fn=MagicMock() + ) is False + + def test_exec_command_python_suppressed(self): + assert render_terminal_preview( + "python3 analyse.py", self._result("result: 42"), print_fn=MagicMock() + ) is False + + def test_exec_command_bash_suppressed(self): + assert render_terminal_preview( + "bash run.sh", self._result("done"), print_fn=MagicMock() + ) is False + + def test_empty_output_returns_false(self): + assert render_terminal_preview("cat foo.py", self._result(""), print_fn=MagicMock()) is False + + def test_invalid_result_json_returns_false(self): + assert render_terminal_preview("cat foo.py", "not json", print_fn=MagicMock()) is False + + +# --------------------------------------------------------------------------- +# Cute-message deduplication (_code_highlight_active) +# --------------------------------------------------------------------------- + +class TestCuteMessageDedup: + def teardown_method(self): + # Always restore flag to default after each test + set_code_highlight_active(False) + + def test_no_snippet_when_active(self): + set_code_highlight_active(True) + msg = get_cute_tool_message("execute_code", {"code": "x = compute()\nreturn x"}, 1.5) + assert "x = compute()" not in msg + assert "return x" not in msg + + def test_snippet_present_when_inactive(self): + set_code_highlight_active(False) + msg = get_cute_tool_message("execute_code", {"code": "x = compute()"}, 1.5) + assert "x = compute()" in msg + + def test_duration_always_present(self): + for active in (True, False): + set_code_highlight_active(active) + msg = get_cute_tool_message("execute_code", {"code": "pass"}, 2.3) + assert "2.3s" in msg + + def test_other_tools_unaffected_by_flag(self): + set_code_highlight_active(True) + msg = get_cute_tool_message("terminal", {"command": "ls -la"}, 0.5) + assert "ls -la" in msg + + +# --------------------------------------------------------------------------- +# _result_succeeded gate (guards execute_code preview on error) +# --------------------------------------------------------------------------- + +class TestHighlightTruncation: + """_emit_highlighted_lines truncates at _HIGHLIGHT_MAX_LINES.""" + + def _collect(self, content: str, language: str = "python") -> list[str]: + lines = [] + _highlight_block("test", content, language, print_fn=lines.append) + return lines # includes the header line + + def test_short_content_not_truncated(self): + code = "\n".join(f"x = {i}" for i in range(10)) + out = self._collect(code) + assert not any("omitted" in l for l in out) + _ansi = re.compile(r"\x1b\[[0-9;]*m") + assert any("x = 9" in _ansi.sub("", l) for l in out) + + def test_long_content_truncated(self): + code = "\n".join(f"x = {i}" for i in range(_HIGHLIGHT_MAX_LINES + 20)) + out = self._collect(code) + assert any("omitted" in l for l in out) + # Should not contain lines beyond the cap + assert not any(f"x = {_HIGHLIGHT_MAX_LINES + 1}" in l for l in out) + + def test_omission_footer_shows_correct_count(self): + extra = 15 + code = "\n".join(f"x = {i}" for i in range(_HIGHLIGHT_MAX_LINES + extra)) + out = self._collect(code) + footer = next(l for l in out if "omitted" in l) + assert str(extra) in footer + + def test_exactly_at_limit_not_truncated(self): + code = "\n".join(f"x = {i}" for i in range(_HIGHLIGHT_MAX_LINES)) + out = self._collect(code) + assert not any("omitted" in l for l in out) + + def test_execute_code_preview_truncates(self): + code = "\n".join(f"x = {i}" for i in range(_HIGHLIGHT_MAX_LINES + 10)) + lines = [] + render_execute_code_preview(code, print_fn=lines.append) + assert any("omitted" in l for l in lines) + + +class TestResultSucceededGate: + def test_error_status_fails(self): + assert not _result_succeeded('{"status": "error", "error": "SyntaxError"}') + + def test_ok_status_passes(self): + assert _result_succeeded('{"status": "ok", "output": "48\\n"}') + + def test_explicit_error_key_fails(self): + assert not _result_succeeded('{"error": "something went wrong"}') + + def test_success_false_fails(self): + assert not _result_succeeded('{"success": false}') + + def test_success_true_passes(self): + assert _result_succeeded('{"success": true}') + + def test_invalid_json_fails(self): + assert not _result_succeeded("not json") + + def test_none_fails(self): + assert not _result_succeeded(None) diff --git a/tests/test_rich_output.py b/tests/test_rich_output.py new file mode 100644 index 0000000000000..edc56fa0960c3 --- /dev/null +++ b/tests/test_rich_output.py @@ -0,0 +1,958 @@ +"""Tests for agent/rich_output.py — syntax highlighting, diff rendering, code block detection.""" + +import re + +import pytest +from unittest.mock import patch + +from agent.rich_output import ( + DiffRenderer, + FilePathFormatter, + LanguageDetector, + StreamingCodeBlockHighlighter, + SyntaxHighlighter, + _intra_diff, + _parse_diff_filename, + apply_block_line, + apply_inline_markdown, + clean_command_output, + format_response, +) + + +# --------------------------------------------------------------------------- +# Shared test helpers +# --------------------------------------------------------------------------- + +def _seg_color(seg) -> str: + """Return the colour name of a Text segment as a plain string.""" + return seg.style.color.name # rich.color.Color.name is the canonical name string + + +def _renderables(diff: str) -> list: + """Return the list of Text children from DiffRenderer._style(). + + Uses ``group.renderables`` which is an internal Rich attribute. If a + future Rich upgrade removes it, switch to rendering the Group to a Console + buffer and inspecting the output instead. + """ + return list(DiffRenderer()._style(diff.splitlines()).renderables) + + +# --------------------------------------------------------------------------- +# LanguageDetector +# --------------------------------------------------------------------------- + +class TestLanguageDetector: + def setup_method(self): + self.det = LanguageDetector() + + def test_detect_python_from_extension(self): + assert self.det.detect_from_filename("foo.py") == "python" + + def test_detect_typescript_from_extension(self): + assert self.det.detect_from_filename("app.ts") == "typescript" + + def test_detect_unknown_extension_returns_none(self): + assert self.det.detect_from_filename("file.xyz") is None + + def test_detect_dockerfile(self): + assert self.det.detect_from_filename("Dockerfile") == "dockerfile" + + def test_detect_makefile(self): + assert self.det.detect_from_filename("Makefile") == "makefile" + + def test_detect_python_from_content(self): + code = "def hello():\n return 42\n" + assert self.det.detect_from_content(code) == "python" + + def test_detect_javascript_from_content(self): + code = "const x = require('fs');\nmodule.exports = x;\n" + assert self.det.detect_from_content(code) == "javascript" + + def test_detect_bash_from_content(self): + code = "#!/bin/bash\nif [ -f foo ]; then echo hi; fi\n" + assert self.det.detect_from_content(code) == "bash" + + def test_detect_empty_content_returns_none(self): + assert self.det.detect_from_content("") is None + assert self.det.detect_from_content(" \n ") is None + + def test_detect_prefers_filename_over_content(self): + # .js extension should win even if content looks like Python + assert self.det.detect("def foo(): pass", filename="script.js") == "javascript" + + +# --------------------------------------------------------------------------- +# FilePathFormatter +# --------------------------------------------------------------------------- + +class TestFilePathFormatter: + def test_python_icon(self): + assert FilePathFormatter.get_file_icon("main.py") == "🐍" + + def test_rust_icon(self): + assert FilePathFormatter.get_file_icon("lib.rs") == "🦀" + + def test_unknown_extension_fallback(self): + assert FilePathFormatter.get_file_icon("weird.xyz") == "📄" + + def test_titled_includes_icon_and_path(self): + result = FilePathFormatter.titled("main.py", compact=False) + assert "🐍" in result + assert "main.py" in result + + def test_format_path_compact_returns_relative(self, tmp_path): + file_path = str(tmp_path / "sub" / "foo.py") + result = FilePathFormatter.format_path(file_path, compact=True, cwd=str(tmp_path)) + assert result == "sub/foo.py" + + def test_format_path_verbose_returns_full(self, tmp_path): + file_path = str(tmp_path / "foo.py") + result = FilePathFormatter.format_path(file_path, compact=False) + assert result == file_path + + +# --------------------------------------------------------------------------- +# SyntaxHighlighter +# --------------------------------------------------------------------------- + +class TestSyntaxHighlighter: + def setup_method(self): + self.hl = SyntaxHighlighter() + + def test_to_ansi_returns_string(self): + result = self.hl.to_ansi("x = 1", language="python") + assert isinstance(result, str) + assert "x" in result + + def test_to_ansi_contains_ansi_codes(self): + result = self.hl.to_ansi("def foo(): pass", language="python") + # Should contain at least one ANSI escape sequence + assert "\033[" in result + + def test_to_markup_returns_string(self): + result = self.hl.to_markup("x = 1", language="python") + assert isinstance(result, str) + + def test_to_ansi_empty_string(self): + result = self.hl.to_ansi("") + assert isinstance(result, str) + + def test_to_ansi_fallback_on_unknown_language(self): + # Unknown language should not crash + result = self.hl.to_ansi("some text", language="nonexistentlang123") + assert isinstance(result, str) + assert "some text" in result + + +# --------------------------------------------------------------------------- +# DiffRenderer +# --------------------------------------------------------------------------- + +class TestDiffRenderer: + def setup_method(self): + self.dr = DiffRenderer() + + def test_to_lines_returns_list(self): + diff = "--- a/foo.py\n+++ b/foo.py\n@@ -1 +1 @@\n-old\n+new\n" + lines = self.dr.to_lines(diff) + assert isinstance(lines, list) + assert len(lines) > 0 + + def test_to_lines_contains_content(self): + diff = "--- a/foo.py\n+++ b/foo.py\n@@ -1 +1 @@\n-old\n+new\n" + lines = self.dr.to_lines(diff) + combined = "\n".join(lines) + assert "old" in combined + assert "new" in combined + + def test_from_content_produces_renderable(self): + from rich.console import Group + result = self.dr.from_content("old line\n", "new line\n", file_path="test.py") + assert isinstance(result, Group) + + def test_from_unified_empty_diff(self): + # Empty diff should not crash + result = self.dr.to_lines("") + assert isinstance(result, list) + + def test_file_header_formatted(self): + diff = "--- a/src/main.py\n+++ b/src/main.py\n@@ -1 +1 @@\n-x\n+y\n" + lines = self.dr.to_lines(diff) + combined = "\n".join(lines) + assert "main.py" in combined + + def test_to_lines_does_not_crash_on_malformed_diff(self): + result = self.dr.to_lines("not a real diff at all\njust some text\n") + assert isinstance(result, list) + + +# --------------------------------------------------------------------------- +# StreamingCodeBlockHighlighter +# --------------------------------------------------------------------------- + +class TestStreamingCodeBlockHighlighter: + def setup_method(self): + self.hl = StreamingCodeBlockHighlighter() + + def test_plain_lines_pass_through(self): + assert self.hl.process_line("Hello world") == "Hello world" + assert self.hl.process_line("Another line") == "Another line" + + def test_opening_fence_suppressed(self): + assert self.hl.process_line("```python") is None + + def test_code_lines_buffered(self): + self.hl.process_line("```python") + assert self.hl.process_line("x = 1") is None + assert self.hl.process_line("y = 2") is None + + def test_closing_fence_flushes_highlighted(self): + self.hl.process_line("```python") + self.hl.process_line("x = 1") + result = self.hl.process_line("```") + assert result is not None + assert "x" in result + + def test_full_code_block_sequence(self): + lines = ["Here is code:", "```python", "def foo(): pass", "```", "Done."] + outputs = [] + for line in lines: + out = self.hl.process_line(line) + if out is not None: + outputs.append(out) + tail = self.hl.flush() + if tail: + outputs.append(tail) + + combined = "\n".join(outputs) + assert "Here is code:" in combined + assert "foo" in combined + assert "Done." in combined + + def test_flush_returns_none_when_no_open_block(self): + assert self.hl.flush() is None + + def test_flush_returns_content_for_unclosed_block(self): + self.hl.process_line("```python") + self.hl.process_line("x = 1") + result = self.hl.flush() + assert result is not None + assert "x" in result + + def test_reset_clears_state(self): + self.hl.process_line("```python") + self.hl.process_line("x = 1") + self.hl.reset() + assert self.hl.flush() is None + # Should behave as fresh after reset + assert self.hl.process_line("normal line") == "normal line" + + def test_multiple_blocks_in_sequence(self): + lines = [ + "Block one:", "```python", "a = 1", "```", + "Block two:", "```javascript", "var b = 2;", "```", + ] + outputs = [self.hl.process_line(l) for l in lines] + non_none = [o for o in outputs if o is not None] + assert len(non_none) == 4 # "Block one:", highlighted, "Block two:", highlighted + + def test_no_language_hint_still_works(self): + self.hl.process_line("```") + self.hl.process_line("SELECT * FROM users;") + result = self.hl.process_line("```") + assert result is not None + assert "SELECT" in result + + def test_lang_hint_passed_to_highlighter(self): + """Opening fence ```python should call to_ansi with language='python'.""" + with patch.object(self.hl._hl, "to_ansi", return_value="highlighted") as mock_ansi: + self.hl.process_line("```python") + self.hl.process_line("x = 1") + self.hl.process_line("```") + mock_ansi.assert_called_once() + _, kwargs = mock_ansi.call_args + assert kwargs.get("language") == "python" + + def test_no_lang_hint_calls_content_detection(self): + """Opening fence with no hint should fall back to detect_from_content.""" + with patch.object(self.hl._det, "detect_from_content", return_value=None) as mock_det: + self.hl.process_line("```") + self.hl.process_line("x = 1") + self.hl.process_line("```") + mock_det.assert_called_once() + + def test_four_backtick_fence_opened_and_closed(self): + """4-backtick opening fence is handled; 4-backtick closing fence closes it.""" + assert self.hl.process_line("````python") is None # suppressed + assert self.hl.process_line("x = 1") is None # buffered + result = self.hl.process_line("````") # closes + assert result is not None + assert "x" in result + + def test_four_backtick_fence_three_backtick_close_ignored(self): + """3-backtick closing fence inside a 4-backtick block is buffered, not a close.""" + assert self.hl.process_line("````python") is None + assert self.hl.process_line("x = 1") is None + # 3-backtick closer must NOT close a 4-backtick block + assert self.hl.process_line("```") is None # still buffering + result = self.hl.flush() # force flush + assert result is not None + assert "x" in result + + def test_prose_after_four_backtick_block_rendered(self): + """Lines after a properly-closed 4-backtick block are treated as prose.""" + self.hl.process_line("````python") + self.hl.process_line("x = 1") + self.hl.process_line("````") + # Back in prose mode — next line should pass through unchanged + out = self.hl.process_line("**bold**") + assert out == "**bold**" + + +# --------------------------------------------------------------------------- +# format_response +# --------------------------------------------------------------------------- + +class TestFormatResponse: + def test_plain_text_unchanged(self): + text = "No code here, just text." + result = format_response(text) + assert "No code here, just text." in result + + def test_code_block_highlighted(self): + text = "Here:\n```python\ndef foo(): pass\n```\nDone." + result = format_response(text) + assert "foo" in result + assert "Here:" in result + assert "Done." in result + + def test_multiple_code_blocks(self): + text = "First:\n```python\nx = 1\n```\nSecond:\n```javascript\nvar y = 2;\n```" + result = format_response(text) + assert "x" in result + assert "y" in result + + def test_no_code_blocks_returns_original(self): + text = "Just a response with no fences." + assert format_response(text) == text + + def test_empty_string(self): + assert format_response("") == "" + + def test_code_block_without_language(self): + text = "```\nSELECT * FROM t;\n```" + result = format_response(text) + assert "SELECT" in result + + def test_no_lang_hint_calls_content_detection(self): + """Fence with no lang tag should call LanguageDetector.detect_from_content.""" + from unittest.mock import patch, MagicMock + with patch("agent.rich_output.LanguageDetector") as MockLD: + mock_instance = MagicMock() + mock_instance.detect_from_content.return_value = None + MockLD.return_value = mock_instance + format_response("```\nSELECT * FROM t;\n```") + mock_instance.detect_from_content.assert_called_once() + + def test_code_block_content_not_markdown_rendered(self): + """Code fence content must not have apply_block_line/apply_inline_markdown applied. + + Pygments plain-text lexer emits some lines without ANSI codes; those + lines must still be skipped by pass 2 so markdown markers stay literal. + """ + text = "```\n### raw heading\n**raw bold**\n- raw item\n```\nDone." + result = format_response(text) + plain = re.sub(r"\x1b\[[0-9;]*m", "", result) + # Code block content must appear literally + assert "### raw heading" in plain + assert "**raw bold**" in plain + assert "- raw item" in plain + # Prose after the block still renders + assert "Done." in plain + + def test_four_backtick_fence_consumed(self): + """format_response consumes 4-backtick fences and highlights their content.""" + text = "Intro.\n````python\nx = 1\n````\nDone." + result = format_response(text) + assert "Intro." in result + assert "Done." in result + assert "x" in result + # Fences should be consumed (no raw backtick-only lines) + for line in result.splitlines(): + assert not line.strip().startswith("````"), f"fence leaked: {line!r}" + + def test_nested_three_in_four_backtick_fence(self): + """3-backtick inner content inside a 4-backtick fence is highlighted as code.""" + text = "````markdown\n```python\ndef foo(): pass\n```\n````\nAfter." + result = format_response(text) + assert "After." in result + # The outer 4-backtick block is consumed; inner ``` lines are code content + for line in result.splitlines(): + assert not line.strip() == "````", f"4-backtick fence leaked: {line!r}" + + def test_code_fence_inside_blockquote_not_consumed(self): + """Fences prefixed with > must not be treated as code block openers. + + Regression: the old un-anchored regex matched ``` mid-line, causing + > ```python lines to be syntax-highlighted and the \x1b guard to then + skip apply_block_line — so blockquote lines rendered with raw > instead + of the ▌ gutter. + """ + text = "> ```python\n> x = 1\n> ```\nAfter." + result = format_response(text) + plain = re.sub(r"\x1b\[[0-9;]*m", "", result) + # Blockquote gutter must appear; raw > must not lead these lines + assert "▌" in plain + assert "After." in plain + +# --------------------------------------------------------------------------- +# clean_command_output +# --------------------------------------------------------------------------- + +class TestCleanCommandOutput: + def test_strips_venv_paths(self): + noisy = "/home/user/venv/lib/python3.11/site-packages/foo.py\nActual output" + result = clean_command_output(noisy) + assert "site-packages" not in result + assert "Actual output" in result + + def test_keeps_meaningful_lines(self): + output = "Build succeeded\n3 tests passed\nDone." + result = clean_command_output(output) + assert "Build succeeded" in result + assert "3 tests passed" in result + + def test_empty_string(self): + assert clean_command_output("") == "" + + def test_removes_excessive_blank_lines(self): + output = "line1\n\n\n\n\nline2" + result = clean_command_output(output) + assert result.count("\n") < 3 + + +# --------------------------------------------------------------------------- +# _intra_diff unit tests +# --------------------------------------------------------------------------- + +class TestIntraDiff: + def test_equal_spans_use_base_colour(self): + del_segs, add_segs = _intra_diff("abc", "abc") + for seg in del_segs + add_segs: + assert not seg.style.bold + assert _seg_color(seg) == "white" + + def test_changed_span_highlighted(self): + del_segs, add_segs = _intra_diff("foo bar", "foo baz") + # There must be at least one bright_red segment in del and bright_green in add + del_highlighted = [s for s in del_segs if _seg_color(s) == "bright_red"] + add_highlighted = [s for s in add_segs if _seg_color(s) == "bright_green"] + assert del_highlighted, "expected at least one bright_red segment in del_segs" + assert add_highlighted, "expected at least one bright_green segment in add_segs" + # All highlighted segments must be bold + assert all(s.style.bold for s in del_highlighted) + assert all(s.style.bold for s in add_highlighted) + # Equal spans must be white and not bold + del_equal = [s for s in del_segs if _seg_color(s) == "white"] + assert del_equal, "expected equal (white) segments in del_segs" + assert all(not s.style.bold for s in del_equal) + + def test_delete_opcode_no_add_seg(self): + del_segs, add_segs = _intra_diff("abcXYZ", "abc") + del_plain = "".join(s.plain for s in del_segs) + add_plain = "".join(s.plain for s in add_segs) + assert "XYZ" in del_plain + assert len(add_plain) == 3 # only "abc" + + def test_insert_opcode_no_del_seg(self): + del_segs, add_segs = _intra_diff("abc", "abcXYZ") + del_plain = "".join(s.plain for s in del_segs) + add_plain = "".join(s.plain for s in add_segs) + assert "XYZ" in add_plain + assert len(del_plain) == 3 # only "abc" + + +# --------------------------------------------------------------------------- +# _parse_diff_filename unit tests +# --------------------------------------------------------------------------- + +class TestParseDiffFilename: + def test_strips_b_prefix(self): + assert _parse_diff_filename("b/src/foo.py") == "foo.py" + + def test_strips_a_prefix(self): + assert _parse_diff_filename("a/src/foo.py") == "foo.py" + + def test_bare_path(self): + assert _parse_diff_filename("path/bar.py") == "bar.py" + + def test_devnull_falls_back_to_from(self): + assert _parse_diff_filename("/dev/null", "a/old.py") == "old.py" + + def test_devnull_no_fallback_returns_question(self): + assert _parse_diff_filename("/dev/null") == "?" + + +# --------------------------------------------------------------------------- +# DiffRenderer v2 rendering tests +# --------------------------------------------------------------------------- + +_SIMPLE_DIFF = ( + "--- a/foo.py\n" + "+++ b/foo.py\n" + "@@ -1,2 +1,2 @@\n" + "-foo bar\n" + "+foo baz\n" + " context\n" +) + +_LOW_RATIO_DIFF = ( + "--- a/foo.py\n" + "+++ b/foo.py\n" + "@@ -1 +1 @@\n" + "-aaaa\n" + "+zzzz\n" +) + + +class TestDiffRendererV2: + def test_intra_diff_skipped_below_ratio(self): + import re + from io import StringIO + from rich.console import Console + buf = StringIO() + Console(file=buf, force_terminal=True, highlight=False, width=220).print( + DiffRenderer()._style(_LOW_RATIO_DIFF.splitlines()) + ) + lines = buf.getvalue().splitlines() + del_line = next(l for l in lines if "aaaa" in re.sub(r"\x1b\[[0-9;]*m", "", l)) + # bright_red bold is encoded as \x1b[1;91; — must not appear on a flat-colour line + assert "\x1b[1;91;" not in del_line + + def test_pairing_per_run_not_per_hunk(self): + # Use pairs with ratio > 0.5 so intra-diff triggers. + # "return foo_value" vs "return bar_value": share "return " + "_value" = 13 chars, + # total = 32, ratio = 26/32 ≈ 0.81. + diff = ( + "--- a/f.py\n+++ b/f.py\n" + "@@ -1,3 +1,3 @@\n" + "-return foo_value\n" + "+return bar_value\n" + " context\n" + "-return foo_result\n" + "+return bar_result\n" + ) + import re + from io import StringIO + from rich.console import Console + buf = StringIO() + Console(file=buf, force_terminal=True, highlight=False, width=220).print( + DiffRenderer()._style(diff.splitlines()) + ) + output = buf.getvalue() + # Both pairs should produce intra-highlighted changed chars + assert output.count("\x1b[1;91;") >= 2 # bright_red bold in both del lines + assert output.count("\x1b[1;92;") >= 2 # bright_green bold in both add lines + + def test_alternating_run_flush(self): + # -A +B -C +D with no context between — should pair (-A,+B) and (-C,+D) + diff = ( + "--- a/f.py\n+++ b/f.py\n" + "@@ -1,2 +1,2 @@\n" + "-alpha\n" + "+ALPHA\n" + "-beta\n" + "+BETA\n" + ) + renderables = _renderables(diff) + all_plain = " ".join(r.plain for r in renderables) + assert "alpha" in all_plain + assert "ALPHA" in all_plain + assert "beta" in all_plain + assert "BETA" in all_plain + + def test_unpaired_lines_flat_colour(self): + diff = ( + "--- a/f.py\n+++ b/f.py\n" + "@@ -1,2 +1,1 @@\n" + "-first del\n" + "-second del\n" + "+one add\n" + ) + import re + from io import StringIO + from rich.console import Console + buf = StringIO() + Console(file=buf, force_terminal=True, highlight=False, width=220).print( + DiffRenderer()._style(diff.splitlines()) + ) + output = buf.getvalue() + lines = output.splitlines() + second_del = next( + l for l in lines + if "second del" in re.sub(r"\x1b\[[0-9;]*m", "", l) + ) + assert "\x1b[91m" not in second_del # no bright_red on unpaired line + + def test_summary_header_add_only(self): + diff = "--- a/f.py\n+++ b/f.py\n@@ -1 +1 @@\n+new\n" + renderables = _renderables(diff) + plain = renderables[0].plain + assert "Added" in plain + assert "removed" not in plain + + def test_summary_header_remove_only(self): + diff = "--- a/f.py\n+++ b/f.py\n@@ -1 +1 @@\n-old\n" + plain = _renderables(diff)[0].plain + assert "Removed" in plain + assert "Added" not in plain + + def test_summary_header_mixed(self): + diff = "--- a/f.py\n+++ b/f.py\n@@ -1 +1 @@\n-old\n+new\n" + plain = _renderables(diff)[0].plain + assert "Added" in plain + assert "removed" in plain + + def test_summary_header_plural(self): + diff = "--- a/f.py\n+++ b/f.py\n@@ -1 +2 @@\n+line1\n+line2\n" + plain = _renderables(diff)[0].plain + assert "Added 2 lines" in plain + + def test_summary_header_singular(self): + diff = "--- a/f.py\n+++ b/f.py\n@@ -1 +1 @@\n+line1\n" + plain = _renderables(diff)[0].plain + assert "Added 1 line" in plain + assert "lines" not in plain + + def test_summary_header_contains_filename(self): + diff = "--- a/src/foo.py\n+++ b/src/foo.py\n@@ -1 +1 @@\n+x\n" + plain = _renderables(diff)[0].plain + assert "foo.py" in plain + + def test_summary_header_strips_b_prefix(self): + diff = "--- a/path/bar.py\n+++ b/path/bar.py\n@@ -1 +1 @@\n+x\n" + plain = _renderables(diff)[0].plain + assert "bar.py" in plain + assert "b/bar.py" not in plain + + def test_summary_header_bare_path(self): + diff = "--- path/bar.py\n+++ path/bar.py\n@@ -1 +1 @@\n+x\n" + plain = _renderables(diff)[0].plain + assert "bar.py" in plain + + def test_summary_header_devnull_fallback(self): + diff = "--- a/old.py\n+++ /dev/null\n@@ -1 +0,0 @@\n-x\n" + plain = _renderables(diff)[0].plain + assert "old.py" in plain + + def test_multi_file_diff_two_headers(self): + diff = ( + "--- a/one.py\n+++ b/one.py\n@@ -1 +1 @@\n+x\n" + "--- a/two.py\n+++ b/two.py\n@@ -1 +1 @@\n+y\n" + ) + renderables = _renderables(diff) + header_plains = [r.plain for r in renderables if "●" in r.plain] + assert len(header_plains) == 2 + assert any("one.py" in p for p in header_plains) + assert any("two.py" in p for p in header_plains) + + def test_separator_width_matches_header(self): + diff = "--- a/foo.py\n+++ b/foo.py\n@@ -1 +1 @@\n+x\n" + renderables = _renderables(diff) + header = renderables[0] + separator = renderables[1] + assert len(separator.plain) == len(header.plain) + + +# --------------------------------------------------------------------------- +# apply_inline_markdown +# --------------------------------------------------------------------------- + +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") + + +def _strip(s: str) -> str: + """Strip all ANSI escape codes from s.""" + return _ANSI_RE.sub("", s) + + +class TestApplyInlineMarkdown: + def test_bold_double_asterisk(self): + result = apply_inline_markdown("**foo**") + assert "\033[1m" in result + assert "foo" in result + assert "**" not in result + + def test_bold_double_underscore(self): + result = apply_inline_markdown("__foo__") + assert "\033[1m" in result + assert "foo" in result + assert "__" not in result + + def test_italic_single_asterisk(self): + result = apply_inline_markdown("*foo*") + assert "\033[3m" in result + assert "foo" in result + assert result.count("*") == 0 + + def test_italic_single_underscore(self): + result = apply_inline_markdown("_foo_") + assert "\033[3m" in result + assert "foo" in result + assert "_" not in result + + def test_italic_underscore_multi_word(self): + result = apply_inline_markdown("_underline - kinda works_") + assert "\033[3m" in result + assert "underline - kinda works" in result + assert "_" not in result + + def test_underscore_inside_word_ignored(self): + result = apply_inline_markdown("snake_case_var") + assert result == "snake_case_var" + + def test_trailing_underscore_ignored(self): + assert apply_inline_markdown("value_") == "value_" + + def test_leading_underscore_ignored(self): + assert apply_inline_markdown("_private") == "_private" + + def test_backtick_code_span(self): + result = apply_inline_markdown("`foo`") + assert "\033[97m" in result + assert "foo" in result + assert "`" not in result + + def test_strikethrough(self): + result = apply_inline_markdown("~~foo~~") + assert "\033[9m" in result + assert "foo" in result + assert "~~" not in result + + def test_mixed_bold_and_code(self): + result = apply_inline_markdown("**Line 88**: `cdOffset`") + assert "\033[1m" in result # bold applied + assert "\033[97m" in result # code span applied + assert "**" not in result + assert "`" not in result + + def test_asterisks_inside_backtick_untouched(self): + result = apply_inline_markdown("`**not bold**`") + # Content inside code span must not be bold-rendered + assert "\033[1m" not in result + assert "**not bold**" in result + + def test_already_ansi_returned_unchanged(self): + ansi_line = "\033[32mgreen\033[0m" + assert apply_inline_markdown(ansi_line) is ansi_line + + def test_empty_string(self): + assert apply_inline_markdown("") == "" + + def test_plain_text_unchanged(self): + assert apply_inline_markdown("plain text") == "plain text" + + def test_reset_suffix_restored_between_spans(self): + colour = "\033[32m" + result = apply_inline_markdown("**a** and *b*", reset_suffix=colour) + # Each closing reset should be followed by the colour suffix + assert f"\033[0m{colour}" in result + + def test_no_markdown(self): + assert apply_inline_markdown("no markdown here") == "no markdown here" + + def test_em_tag_italic(self): + result = apply_inline_markdown("foo") + assert "\033[3m" in result + assert "foo" in result + assert "" not in result + assert "" not in result + + def test_strong_tag_bold(self): + result = apply_inline_markdown("foo") + assert "\033[1m" in result + assert "foo" in result + assert "" not in result + + def test_link_underlined(self): + result = apply_inline_markdown("[click here](https://x.com)") + assert "\033[4m" in result + assert "click here" in result + assert "https://x.com" in result + assert "[click here]" not in _strip(result) + + def test_image_placeholder(self): + result = apply_inline_markdown("![logo](img.png)") + assert "[img: logo]" in result + assert "\033[2m" in result + assert "img.png" not in result + + def test_image_before_link(self): + result = apply_inline_markdown("![a](u) [b](v)") + assert "[img: a]" in result + assert "\033[4m" in result + assert "b" in result + + +class TestApplyBlockLine: + def test_h1_stripped_and_bold(self): + result = apply_block_line("# Foo") + assert "\033[1;97m" in result + assert "Foo" in result + assert "#" not in result + + def test_h2_dimmer_than_h1(self): + result = apply_block_line("## Foo") + assert "\033[1;37m" in result + assert "97m" not in result + + def test_h4_bold_dim(self): + result = apply_block_line("#### Foo") + assert "\033[1;2m" in result + + def test_h1_with_inline_span(self): + result = apply_block_line("# **Foo**") + assert "\033[1;97m" in result + assert "\033[1m" in result + assert "Foo" in result + assert "**" not in result + + def test_hr_dashes_replaced(self): + result = apply_block_line("---") + assert "─" in result + assert "-" not in _strip(result) + + def test_hr_stars_replaced(self): + result = apply_block_line("***") + assert "─" in result + + def test_hr_underscores_replaced(self): + result = apply_block_line("___") + assert "─" in result + + def test_non_hr_dashes_unchanged(self): + result = apply_block_line("some --- text") + assert result == "some --- text" + + def test_blockquote_gutter(self): + result = apply_block_line("> hello") + assert "▌" in result + assert "hello" in result + assert ">" not in result + + def test_blockquote_nested_collapsed(self): + result = apply_block_line(">> deep") + assert result.count("▌") == 1 + + def test_blockquote_inline_span(self): + result = apply_block_line("> **bold**") + assert "▌" in result + assert "\033[1m" in result + assert "**" not in result + + def test_blockquote_inline_span_restores_dim(self): + # Bold span inside a blockquote must restore the dim gutter style on close, + # not reset to terminal default — fixes missing reset_suffix on blockquote branch. + result = apply_block_line("> **bold** plain") + # Dim style (\033[2m) must appear after the bold close (\033[0m) + assert "\033[0m\033[2m" in result + + def test_list_bullet_dot(self): + result = apply_block_line("- item") + assert "•" in result + assert "item" in result + assert result.startswith("•") + + def test_list_bullet_circle_nested(self): + result = apply_block_line(" - item") + assert "◦" in result + + def test_list_bullet_triangle_double_nested(self): + result = apply_block_line(" - item") + assert "▸" in result + + def test_list_star_and_plus(self): + assert "•" in apply_block_line("* item") + assert "•" in apply_block_line("+ item") + + def test_ordered_list_unchanged(self): + result = apply_block_line("1. item") + assert result == "1. item" + + def test_reference_link_suppressed(self): + result = apply_block_line("[ref]: https://x.com") + assert result == "" + + def test_reference_link_with_quoted_title_suppressed(self): + assert apply_block_line('[ref]: https://x.com "Page Title"') == "" + + def test_reference_link_with_paren_title_suppressed(self): + assert apply_block_line("[ref]: https://x.com (Page Title)") == "" + + def test_ansi_lines_skipped(self): + ansi_line = "\033[32mgreen\033[0m" + assert apply_block_line(ansi_line) is ansi_line + + def test_multiline_skipped(self): + multi = "line1\nline2" + assert apply_block_line(multi) is multi + + def test_plain_line_unchanged(self): + assert apply_block_line("just text") == "just text" + + +class TestFormatResponseInlineMarkdown: + """Integration: format_response applies inline markdown to prose, not code.""" + + def test_bold_in_prose_rendered(self): + text = "This is **important** text." + result = format_response(text) + assert "\033[1m" in result + assert "important" in result + assert "**" not in result + + def test_heading_followed_by_paragraph_preserves_newline(self): + # apply_block_line drops the trailing \n from matched lines; format_response + # must compensate so the paragraph starts on its own line. + text = "# Title\nParagraph text" + result = format_response(text) + plain = _strip(result) + # Heading and paragraph must be on separate lines + assert plain.index("Title") < plain.index("\n") + assert "Paragraph text" in plain + + def test_list_followed_by_paragraph_preserves_newline(self): + text = "- item one\nnext line" + result = format_response(text) + plain = _strip(result) + assert "item one" in plain + assert plain.index("item one") < plain.index("\n") + assert "next line" in plain + + def test_code_block_not_double_escaped(self): + text = "Note **this**:\n```python\nx = **1**\n```\nEnd **here**." + result = format_response(text) + # Prose bold rendered + assert "\033[1m" in result + # The Python code block was syntax-highlighted; the ** inside are code + # content — they appear as plain chars inside the highlighted block, + # not as ANSI bold markers. Verify no double-escape by checking that + # the result does not contain literal \033[1m immediately followed by + # content that was already inside an ANSI span. + # Simpler: strip all ANSI and confirm code content intact + plain = _strip(result) + assert "x = **1**" in plain + + def test_backslash_escape_stripped(self): + r"""CommonMark backslash escapes like \] and \* are stripped from output.""" + result = apply_inline_markdown(r"- [ \] unchecked") + assert r"\]" not in result + assert "]" in result + + def test_backslash_escape_checkbox(self): + r"""[x\] renders as [x] — backslash before ] removed.""" + result = apply_inline_markdown(r"- [x\] checked item") + assert r"\]" not in result + assert "[x]" in _strip(result)