diff --git a/clide/app.py b/clide/app.py index 09a0b9bb..64d9ec3a 100644 --- a/clide/app.py +++ b/clide/app.py @@ -606,7 +606,6 @@ class ClideApp(App[None]): else: self.notify("Undo not available", severity="warning") - def action_quick_open(self) -> None: """Quick file open.""" # TODO: Implement quick open @@ -739,27 +738,6 @@ class ClideApp(App[None]): workspace = self.query_one(WorkspacePanel) workspace.clear_diff() - async def on_workspace_panel_command_submitted( - self, - event: WorkspacePanel.CommandSubmitted, - ) -> None: - """Handle terminal command - run and show output.""" - from clide.services.process_service import ProcessService - - workspace = self.query_one(WorkspacePanel) - result = await ProcessService.run_async( - event.command, - cwd=self.workdir, - shell=True, - ) - if result.stdout: - workspace.write_terminal_output(result.stdout) - if result.stderr: - workspace.write_terminal_error(result.stderr) - - # Refresh after command - await self._refresh_git() - def on_claude_panel_claude_started( self, _event: ClaudePanel.ClaudeStarted, diff --git a/clide/widgets/components/terminal_display.py b/clide/widgets/components/terminal_display.py new file mode 100644 index 00000000..6e5b9719 --- /dev/null +++ b/clide/widgets/components/terminal_display.py @@ -0,0 +1,631 @@ +"""Reusable PTY-based terminal emulator widget using pyte.""" + +from __future__ import annotations + +import asyncio +import codecs +import fcntl +import os +import pty +import re +import struct +import termios +import time + +from rich.text import Text +from textual.strip import Strip +from textual.widget import Widget + +from clide.services.settings_service import get_settings_service + +# Use vendored pyte with diagnostic logging support +from clide.vendor import pyte + + +def _setup_terminal_debug_logging() -> None: + """Set up terminal debug logging if enabled in settings.""" + settings = get_settings_service() + if not settings.get("terminal_debug", False): + return + + # Create log file in settings directory + log_path = settings.settings_dir / "terminal_debug.log" + log_path.parent.mkdir(parents=True, exist_ok=True) + + # Open log file (append mode) + log_file = open(log_path, "a", encoding="utf-8") + + def debug_logger(message: str) -> None: + """Log debug message with timestamp.""" + import datetime + + timestamp = datetime.datetime.now().isoformat() + log_file.write(f"[{timestamp}] {message}\n") + log_file.flush() + + # Set up pyte debug logging + pyte.set_debug_logger(debug_logger) + + +class TerminalDisplay(Widget, can_focus=True): + """A terminal emulator widget using pyte.""" + + DEFAULT_CSS = """ + TerminalDisplay { + height: 100%; + width: 100%; + background: $background; + } + """ + + # Internal padding (rendered as part of terminal content, uses terminal bg) + PADDING_LEFT = 1 + PADDING_RIGHT = 1 + + def __init__( + self, + cols: int = 80, + rows: int = 24, + **kwargs, + ) -> None: + super().__init__(**kwargs) + self._cols = cols + self._rows = rows + + # Set up debug logging before creating pyte objects + _setup_terminal_debug_logging() + + self._screen = pyte.HistoryScreen(cols, rows, history=1000) + self._screen.set_mode(pyte.modes.LNM) # Line feed mode + self._stream = pyte.Stream(self._screen) + self._master_fd: int | None = None + self._pid: int | None = None + self._read_task: asyncio.Task | None = None + self._refresh_task: asyncio.Task | None = None + self._needs_refresh: bool = False + self._last_refresh: float = 0 + self._pending_start: tuple[str, str] | None = None + # Incremental UTF-8 decoder to handle partial sequences at buffer boundaries + self._decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + # Scroll offset for viewing history (0 = at bottom/current, positive = scrolled up) + self._scroll_offset: int = 0 + + def on_mount(self) -> None: + """Initialize terminal size from widget dimensions.""" + # Get actual widget size, accounting for internal padding + size = self.size + if size.width > 0 and size.height > 0: + self._cols = max(size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20) + self._rows = size.height + self._screen.resize(self._rows, self._cols) + + # If start was called before mount, do it now + if self._pending_start: + command, cwd = self._pending_start + self._pending_start = None + self._do_start(command, cwd) + + def on_resize(self, event) -> None: + """Handle terminal resize.""" + # Get new size from widget, accounting for internal padding + new_cols = max(self.size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20) + new_rows = max(self.size.height, 5) + + if new_cols != self._cols or new_rows != self._rows: + old_cols = self._cols + + # Update dimensions first + self._cols = new_cols + self._rows = new_rows + + # Resize pyte screen - this will preserve content where possible + self._screen.resize(new_rows, new_cols) + + # If screen got wider, clear the new columns to avoid stale data + # pyte's resize should handle this, but let's be defensive + if new_cols > old_cols: + for y in range(new_rows): + line = self._screen.buffer[y] + for x in range(old_cols, new_cols): + # Clear any stale data in new columns + line[x] = pyte.screens.Char(" ") + + # Update PTY size if running - child process will redraw + if self._master_fd is not None: + self._set_pty_size(self._master_fd, new_rows, new_cols) + + # Force a full refresh + self.refresh() + + def _set_pty_size(self, fd: int, rows: int, cols: int) -> None: + """Set the PTY window size and notify the child process.""" + try: + winsize = struct.pack("HHHH", rows, cols, 0, 0) + fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize) + + # Send SIGWINCH to notify the child process of resize + if self._pid is not None: + import signal + + try: + os.kill(self._pid, signal.SIGWINCH) + except OSError: + pass + except OSError: + pass + + def start(self, command: str, cwd: str) -> None: + """Start a process in the terminal.""" + # If not mounted yet, defer start + if not self.is_mounted: + self._pending_start = (command, cwd) + return + + self._do_start(command, cwd) + + def _do_start(self, command: str, cwd: str) -> None: + """Actually start the process in the terminal.""" + # Reset decoder state for new process + self._decoder.reset() + + # Get current widget size, accounting for internal padding + size = self.size + if size.width > 0 and size.height > 0: + self._cols = max(size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20) + self._rows = size.height + self._screen.resize(self._rows, self._cols) + + # Fork a PTY + pid, master_fd = pty.fork() + + if pid == 0: + # Child process + os.chdir(cwd) + os.environ["TERM"] = "xterm-256color" + os.environ["COLORTERM"] = "truecolor" + os.environ["COLUMNS"] = str(self._cols) + os.environ["LINES"] = str(self._rows) + os.execlp(command, command) + else: + # Parent process + self._pid = pid + self._master_fd = master_fd + + # Set non-blocking + flags = fcntl.fcntl(master_fd, fcntl.F_GETFL) + fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) + + # Set initial size + self._set_pty_size(master_fd, self._rows, self._cols) + + # Start reading + self._read_task = asyncio.create_task(self._read_output()) + + def _schedule_refresh(self) -> None: + """Schedule a throttled refresh to avoid glitching.""" + self._needs_refresh = True + if self._refresh_task is None or self._refresh_task.done(): + self._refresh_task = asyncio.create_task(self._throttled_refresh()) + + async def _throttled_refresh(self) -> None: + """Refresh at most every 16ms (~60fps) to avoid glitching.""" + min_interval = 0.016 # ~60fps + while self._needs_refresh: + now = time.monotonic() + elapsed = now - self._last_refresh + if elapsed < min_interval: + await asyncio.sleep(min_interval - elapsed) + self._needs_refresh = False + self._last_refresh = time.monotonic() + self.refresh() + # Small delay to batch rapid updates + await asyncio.sleep(0.008) + + # Regex to filter escape sequences that pyte doesn't handle + # Kitty keyboard protocol, bracketed paste mode queries, etc. + _UNSUPPORTED_ESCAPES = re.compile( + r"\x1b\[[\=\>\<][0-9;]*[a-zA-Z]" # Kitty keyboard protocol (=, >, or < prefix) + r"|\x1b\[\?[0-9;]*u" # Kitty keyboard query + r"|\x1b\[\?[0-9;]*c" # Device attributes query + r"|\x1b\[>[0-9;]*c" # Secondary device attributes + r"|\x1b\]\d+;[^\x07\x1b]*(?:\x07|\x1b\\)" # OSC sequences (title, etc.) + r"|\x1b\[\?2026[hl]" # Synchronized update mode (not used by pyte) + ) + + def _filter_unsupported_escapes(self, data: str) -> str: + """Filter out escape sequences that pyte doesn't handle.""" + return self._UNSUPPORTED_ESCAPES.sub("", data) + + async def _read_output(self) -> None: + """Read output from the PTY.""" + if self._master_fd is None: + return + + while True: + try: + # Wait for data to be available + await asyncio.sleep(0.005) + + try: + data = os.read(self._master_fd, 65536) + if not data: + break + + # Use incremental decoder to handle partial UTF-8 sequences + # at buffer boundaries (prevents box-drawing chars getting corrupted) + text = self._decoder.decode(data) + if not text: + continue # Still waiting for more bytes to complete a sequence + text = self._filter_unsupported_escapes(text) + + # Feed data to pyte + self._stream.feed(text) + self._schedule_refresh() + + except BlockingIOError: + # No data available + continue + except OSError: + # PTY closed + break + + except asyncio.CancelledError: + break + + def stop(self) -> None: + """Stop the terminal process.""" + if self._read_task: + self._read_task.cancel() + self._read_task = None + + if self._refresh_task: + self._refresh_task.cancel() + self._refresh_task = None + + if self._master_fd is not None: + try: + os.close(self._master_fd) + except OSError: + pass + self._master_fd = None + + if self._pid is not None: + try: + os.kill(self._pid, 9) + os.waitpid(self._pid, 0) + except (OSError, ChildProcessError): + pass + self._pid = None + + def is_running(self) -> bool: + """Check if the process is still running.""" + if self._pid is None: + return False + + try: + pid, status = os.waitpid(self._pid, os.WNOHANG) + if pid == 0: + return True # Still running + else: + self._pid = None + return False + except ChildProcessError: + self._pid = None + return False + + def send(self, data: str) -> None: + """Send data to the terminal.""" + if self._master_fd is not None: + try: + os.write(self._master_fd, data.encode("utf-8")) + except OSError: + pass + + # ANSI 256-color palette (standard 16 colors) + ANSI_COLORS = { + "black": "#000000", + "red": "#cd0000", + "green": "#00cd00", + "yellow": "#cdcd00", + "blue": "#0000ee", + "magenta": "#cd00cd", + "cyan": "#00cdcd", + "white": "#e5e5e5", + "brightblack": "#7f7f7f", + "brightred": "#ff0000", + "brightgreen": "#00ff00", + "brightyellow": "#ffff00", + "brightblue": "#5c5cff", + "brightmagenta": "#ff00ff", + "brightcyan": "#00ffff", + "brightwhite": "#ffffff", + } + + def _convert_color(self, color: str, is_bg: bool = False) -> str | None: + """Convert pyte color to Rich color string.""" + if not color or color == "default": + return None + + # Handle hex colors (with or without # prefix) + if color.startswith("#"): + return color + + # Check if it's a hex color without # (pyte returns "ff0000" not "#ff0000") + if len(color) == 6 and all(c in "0123456789abcdefABCDEF" for c in color): + return f"#{color}" + + # Handle named ANSI colors + color_lower = color.lower() + if color_lower in self.ANSI_COLORS: + return self.ANSI_COLORS[color_lower] + + # Handle 256-color palette (numeric) + try: + num = int(color) + if 0 <= num <= 255: + return f"color({num})" + except ValueError: + pass + + # Fallback: try to use the color name directly + return color + + def _get_line_at(self, y: int): + """Get the line at display position y, accounting for scroll offset. + + When scrolled, we show lines from history mixed with current buffer. + scroll_offset=0 means showing current screen. + scroll_offset=N means the top of display shows N lines back in history. + """ + history = self._screen.history.top + history_len = len(history) + + if self._scroll_offset == 0: + # Not scrolled - show current buffer + return self._screen.buffer[y] + + # Calculate which line to show + # Display line 0 should show history[history_len - scroll_offset] + # Display line N should show history[history_len - scroll_offset + N] + # If that index >= history_len, we're into the current buffer + + history_index = history_len - self._scroll_offset + y + + if history_index < 0: + # Before start of history - return blank + return None + elif history_index < history_len: + # In history + return history[history_index] + else: + # In current buffer + buffer_index = history_index - history_len + if buffer_index < self._rows: + return self._screen.buffer[buffer_index] + return None + + def render_line(self, y: int) -> Strip: + """Render a line of the terminal.""" + # Screen buffer dimensions (what pyte has) + screen_cols = self._cols + screen_rows = self._rows + + # Widget display dimensions (what we need to output) + output_width = ( + self.size.width + if self.size.width > 0 + else screen_cols + self.PADDING_LEFT + self.PADDING_RIGHT + ) + + if y >= screen_rows: + return Strip.blank(output_width) + + # Get the line to render, accounting for scroll offset + line = self._get_line_at(y) + if line is None: + return Strip.blank(output_width) + + text = Text() + + # Add left padding (uses terminal background) + text.append(" " * self.PADDING_LEFT) + + # Only access indices within the screen buffer + cols_to_render = min(screen_cols, output_width - self.PADDING_LEFT - self.PADDING_RIGHT) + + # Debug logging for render pipeline + debug_logger = pyte.get_debug_logger() + log_this_line = False + + for x in range(cols_to_render): + char = line[x] + char_data = char.data if char.data else " " + + # Check for box-drawing and other potentially problematic characters + if len(char_data) == 1: + code = ord(char_data) + # Log box-drawing characters (U+2500-U+257F) + if 0x2500 <= code <= 0x257F: + log_this_line = True + if debug_logger: + debug_logger( + f"RENDER y={y} x={x}: box-drawing U+{code:04X} char='{char_data}'" + ) + + # Handle characters that may not render correctly + if len(char_data) == 1: + code = ord(char_data) + # Control characters (except space) + if code < 32 and code != 0 or 127 <= code <= 159: + char_data = " " + # Braille patterns (U+2800-U+28FF) - used for spinners + # Replace with simple ASCII spinner chars or spaces + elif 0x2800 <= code <= 0x28FF: + # Map braille spinner to simple dots + char_data = "·" + + # Build style from pyte character attributes + style_parts = [] + + # Foreground color + fg = self._convert_color(char.fg) + if fg: + style_parts.append(fg) + + # Background color + bg = self._convert_color(char.bg, is_bg=True) + if bg: + style_parts.append(f"on {bg}") + + # Text attributes + if char.bold: + style_parts.append("bold") + if char.italics: + style_parts.append("italic") + if char.underscore: + style_parts.append("underline") + if char.strikethrough: + style_parts.append("strike") + if char.reverse: + style_parts.append("reverse") + + style = " ".join(style_parts) if style_parts else None + text.append(char_data, style=style) + + # Pad with spaces to fill remaining width (includes right padding) + content_width = self.PADDING_LEFT + cols_to_render + if output_width > content_width: + text.append(" " * (output_width - content_width)) + + # Debug: log the Rich Text content before rendering + if log_this_line and debug_logger: + debug_logger(f"RENDER y={y}: Rich Text plain='{text.plain[:80]}...'") + + # Render text to segments for Strip + segments = list(text.render(self.app.console)) + + # Debug: log segments if we had box-drawing chars + if log_this_line and debug_logger: + for i, seg in enumerate(segments[:10]): # First 10 segments + seg_text = seg.text if hasattr(seg, "text") else str(seg) + if len(seg_text) <= 5: + debug_logger(f"RENDER y={y} seg[{i}]: '{seg_text}' (repr: {repr(seg_text)})") + + strip = Strip(segments) + + # Ensure strip is exactly the output width + if strip.cell_length != output_width: + strip = strip.crop_extend(0, output_width, None) + + return strip + + def on_key(self, event) -> None: + """Handle key presses.""" + # Map special keys + key_map = { + "enter": "\r", + "tab": "\t", + "backspace": "\x7f", + "delete": "\x1b[3~", + "up": "\x1b[A", + "down": "\x1b[B", + "right": "\x1b[C", + "left": "\x1b[D", + "home": "\x1b[H", + "end": "\x1b[F", + "pageup": "\x1b[5~", + "pagedown": "\x1b[6~", + "escape": "\x1b", + } + + if event.key in key_map: + self.send(key_map[event.key]) + event.prevent_default() + event.stop() + elif event.key == "ctrl+c": + self.send("\x03") + event.prevent_default() + event.stop() + elif event.key == "ctrl+d": + self.send("\x04") + event.prevent_default() + event.stop() + elif event.key == "ctrl+z": + self.send("\x1a") + event.prevent_default() + event.stop() + elif event.key == "ctrl+l": + self.send("\x0c") + event.prevent_default() + event.stop() + elif event.key == "shift+pageup": + # Scroll up in history + self._scroll_up(self._rows // 2) + event.prevent_default() + event.stop() + elif event.key == "shift+pagedown": + # Scroll down in history + self._scroll_down(self._rows // 2) + event.prevent_default() + event.stop() + elif event.key == "shift+home": + # Scroll to top of history + self._scroll_to_top() + event.prevent_default() + event.stop() + elif event.key == "shift+end": + # Scroll to bottom (current) + self._scroll_to_bottom() + event.prevent_default() + event.stop() + elif event.character and len(event.character) == 1: + # Any typing scrolls to bottom + self._scroll_to_bottom() + self.send(event.character) + event.prevent_default() + event.stop() + + def on_mouse_scroll_up(self, event) -> None: + """Handle mouse scroll up (view older content).""" + self._scroll_up(3) + event.prevent_default() + event.stop() + + def on_mouse_scroll_down(self, event) -> None: + """Handle mouse scroll down (view newer content).""" + self._scroll_down(3) + event.prevent_default() + event.stop() + + def _scroll_up(self, lines: int) -> None: + """Scroll up (back in history) by given number of lines.""" + max_scroll = len(self._screen.history.top) + new_offset = min(self._scroll_offset + lines, max_scroll) + if new_offset != self._scroll_offset: + self._scroll_offset = new_offset + self.refresh() + + def _scroll_down(self, lines: int) -> None: + """Scroll down (forward toward current) by given number of lines.""" + new_offset = max(self._scroll_offset - lines, 0) + if new_offset != self._scroll_offset: + self._scroll_offset = new_offset + self.refresh() + + def _scroll_to_top(self) -> None: + """Scroll to the top of history.""" + max_scroll = len(self._screen.history.top) + if self._scroll_offset != max_scroll: + self._scroll_offset = max_scroll + self.refresh() + + def _scroll_to_bottom(self) -> None: + """Scroll to the bottom (current screen).""" + if self._scroll_offset != 0: + self._scroll_offset = 0 + self.refresh() + + @property + def history_size(self) -> int: + """Get the number of lines in scrollback history.""" + return len(self._screen.history.top) diff --git a/clide/widgets/components/terminal_pane.py b/clide/widgets/components/terminal_pane.py index badfceb5..6a73f5b2 100644 --- a/clide/widgets/components/terminal_pane.py +++ b/clide/widgets/components/terminal_pane.py @@ -1,15 +1,17 @@ -"""Terminal pane component.""" +"""Terminal pane component with full PTY support.""" +import os from pathlib import Path from textual.app import ComposeResult from textual.containers import Vertical -from textual.message import Message -from textual.widgets import Input, RichLog, Static +from textual.widgets import Static + +from clide.widgets.components.terminal_display import TerminalDisplay class TerminalPane(Vertical): - """Simple terminal/command runner pane.""" + """Full PTY terminal pane running an interactive shell.""" DEFAULT_CSS = """ TerminalPane { @@ -22,46 +24,26 @@ class TerminalPane(Vertical): padding: 0 1; } - TerminalPane RichLog { + TerminalPane TerminalDisplay { height: 1fr; - background: $background; - } - - TerminalPane Input { - dock: bottom; - height: 1; - } - - TerminalPane .prompt { - color: $primary; - } - - TerminalPane .output { - color: $foreground; - } - - TerminalPane .error { - color: $error; } """ - class CommandSubmitted(Message): - """Emitted when a command is submitted.""" - - def __init__(self, command: str) -> None: - self.command = command - super().__init__() - def __init__(self, cwd: Path | None = None, **kwargs) -> None: super().__init__(**kwargs) self._cwd = cwd or Path.cwd() - self._history: list[str] = [] - self._history_index = 0 + self._terminal: TerminalDisplay | None = None + self._shell = os.environ.get("SHELL", "/bin/bash") def compose(self) -> ComposeResult: yield Static(f"Terminal - {self._cwd}", classes="terminal-header") - yield RichLog(id="terminal-log", highlight=True, markup=True) - yield Input(placeholder="Enter command...", id="terminal-input") + self._terminal = TerminalDisplay(id="terminal-display") + yield self._terminal + + def on_mount(self) -> None: + """Start the shell when mounted.""" + if self._terminal: + self._terminal.start(self._shell, str(self._cwd)) @property def cwd(self) -> Path: @@ -75,66 +57,18 @@ class TerminalPane(Vertical): header = self.query_one(".terminal-header", Static) header.update(f"Terminal - {path}") - def write_output(self, text: str, style: str = "output") -> None: - """Write output to terminal. + def focus_terminal(self) -> None: + """Focus the terminal display.""" + if self._terminal: + self._terminal.focus() - Args: - text: Text to write - style: Style class (output, error, prompt) - """ - log = self.query_one("#terminal-log", RichLog) - if style == "error": - log.write(f"[red]{text}[/]") - elif style == "prompt": - log.write(f"[bold cyan]$ {text}[/]") - else: - log.write(text) + def stop(self) -> None: + """Stop the terminal process.""" + if self._terminal: + self._terminal.stop() - def write_command(self, command: str) -> None: - """Write a command to terminal (with prompt).""" - self.write_output(command, "prompt") - - def write_error(self, error: str) -> None: - """Write an error to terminal.""" - self.write_output(error, "error") - - def clear(self) -> None: - """Clear terminal output.""" - log = self.query_one("#terminal-log", RichLog) - log.clear() - - def on_input_submitted(self, event: Input.Submitted) -> None: - """Handle command submission.""" - command = event.value.strip() - if not command: - return - - # Add to history - self._history.append(command) - self._history_index = len(self._history) - - # Clear input - event.input.clear() - - # Write command to output - self.write_command(command) - - # Post message for handling - self.post_message(self.CommandSubmitted(command)) - - def history_up(self) -> None: - """Navigate history up.""" - if self._history and self._history_index > 0: - self._history_index -= 1 - input_widget = self.query_one("#terminal-input", Input) - input_widget.value = self._history[self._history_index] - - def history_down(self) -> None: - """Navigate history down.""" - input_widget = self.query_one("#terminal-input", Input) - if self._history_index < len(self._history) - 1: - self._history_index += 1 - input_widget.value = self._history[self._history_index] - else: - self._history_index = len(self._history) - input_widget.clear() + def restart(self) -> None: + """Restart the shell.""" + if self._terminal: + self._terminal.stop() + self._terminal.start(self._shell, str(self._cwd)) diff --git a/clide/widgets/panels/claude.py b/clide/widgets/panels/claude.py index daab909a..d7b2f5bb 100644 --- a/clide/widgets/panels/claude.py +++ b/clide/widgets/panels/claude.py @@ -3,643 +3,20 @@ from __future__ import annotations import asyncio -import codecs -import fcntl -import os -import pty -import re import shutil -import struct -import termios -import time from pathlib import Path from typing import TYPE_CHECKING -from rich.text import Text from textual.containers import Vertical from textual.message import Message from textual.reactive import reactive -from textual.strip import Strip -from textual.widget import Widget -from clide.services.settings_service import get_settings_service - -# Use vendored pyte with diagnostic logging support -from clide.vendor import pyte +from clide.widgets.components.terminal_display import TerminalDisplay if TYPE_CHECKING: from textual.app import ComposeResult -def _setup_terminal_debug_logging() -> None: - """Set up terminal debug logging if enabled in settings.""" - settings = get_settings_service() - if not settings.get("terminal_debug", False): - return - - # Create log file in settings directory - log_path = settings.settings_dir / "terminal_debug.log" - log_path.parent.mkdir(parents=True, exist_ok=True) - - # Open log file (append mode) - log_file = open(log_path, "a", encoding="utf-8") - - def debug_logger(message: str) -> None: - """Log debug message with timestamp.""" - import datetime - - timestamp = datetime.datetime.now().isoformat() - log_file.write(f"[{timestamp}] {message}\n") - log_file.flush() - - # Set up pyte debug logging - pyte.set_debug_logger(debug_logger) - - -class TerminalDisplay(Widget, can_focus=True): - """A terminal emulator widget using pyte.""" - - DEFAULT_CSS = """ - TerminalDisplay { - height: 100%; - width: 100%; - background: $background; - } - """ - - # Internal padding (rendered as part of terminal content, uses terminal bg) - PADDING_LEFT = 1 - PADDING_RIGHT = 1 - - def __init__( - self, - cols: int = 80, - rows: int = 24, - **kwargs, - ) -> None: - super().__init__(**kwargs) - self._cols = cols - self._rows = rows - - # Set up debug logging before creating pyte objects - _setup_terminal_debug_logging() - - self._screen = pyte.HistoryScreen(cols, rows, history=1000) - self._screen.set_mode(pyte.modes.LNM) # Line feed mode - self._stream = pyte.Stream(self._screen) - self._master_fd: int | None = None - self._pid: int | None = None - self._read_task: asyncio.Task | None = None - self._refresh_task: asyncio.Task | None = None - self._needs_refresh: bool = False - self._last_refresh: float = 0 - self._pending_start: tuple[str, str] | None = None - # Incremental UTF-8 decoder to handle partial sequences at buffer boundaries - self._decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") - # Scroll offset for viewing history (0 = at bottom/current, positive = scrolled up) - self._scroll_offset: int = 0 - - def on_mount(self) -> None: - """Initialize terminal size from widget dimensions.""" - # Get actual widget size, accounting for internal padding - size = self.size - if size.width > 0 and size.height > 0: - self._cols = max(size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20) - self._rows = size.height - self._screen.resize(self._rows, self._cols) - - # If start was called before mount, do it now - if self._pending_start: - command, cwd = self._pending_start - self._pending_start = None - self._do_start(command, cwd) - - def on_resize(self, event) -> None: - """Handle terminal resize.""" - # Get new size from widget, accounting for internal padding - new_cols = max(self.size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20) - new_rows = max(self.size.height, 5) - - if new_cols != self._cols or new_rows != self._rows: - old_cols = self._cols - - # Update dimensions first - self._cols = new_cols - self._rows = new_rows - - # Resize pyte screen - this will preserve content where possible - self._screen.resize(new_rows, new_cols) - - # If screen got wider, clear the new columns to avoid stale data - # pyte's resize should handle this, but let's be defensive - if new_cols > old_cols: - for y in range(new_rows): - line = self._screen.buffer[y] - for x in range(old_cols, new_cols): - # Clear any stale data in new columns - line[x] = pyte.screens.Char(" ") - - # Update PTY size if running - Claude will redraw - if self._master_fd is not None: - self._set_pty_size(self._master_fd, new_rows, new_cols) - - # Force a full refresh - self.refresh() - - def _set_pty_size(self, fd: int, rows: int, cols: int) -> None: - """Set the PTY window size and notify the child process.""" - try: - winsize = struct.pack("HHHH", rows, cols, 0, 0) - fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize) - - # Send SIGWINCH to notify the child process of resize - if self._pid is not None: - import signal - - try: - os.kill(self._pid, signal.SIGWINCH) - except OSError: - pass - except OSError: - pass - - def start(self, command: str, cwd: str) -> None: - """Start a process in the terminal.""" - # If not mounted yet, defer start - if not self.is_mounted: - self._pending_start = (command, cwd) - return - - self._do_start(command, cwd) - - def _do_start(self, command: str, cwd: str) -> None: - """Actually start the process in the terminal.""" - # Reset decoder state for new process - self._decoder.reset() - - # Get current widget size, accounting for internal padding - size = self.size - if size.width > 0 and size.height > 0: - self._cols = max(size.width - self.PADDING_LEFT - self.PADDING_RIGHT, 20) - self._rows = size.height - self._screen.resize(self._rows, self._cols) - - # Fork a PTY - pid, master_fd = pty.fork() - - if pid == 0: - # Child process - os.chdir(cwd) - os.environ["TERM"] = "xterm-256color" - os.environ["COLORTERM"] = "truecolor" - os.environ["COLUMNS"] = str(self._cols) - os.environ["LINES"] = str(self._rows) - os.execlp(command, command) - else: - # Parent process - self._pid = pid - self._master_fd = master_fd - - # Set non-blocking - flags = fcntl.fcntl(master_fd, fcntl.F_GETFL) - fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK) - - # Set initial size - self._set_pty_size(master_fd, self._rows, self._cols) - - # Start reading - self._read_task = asyncio.create_task(self._read_output()) - - def _schedule_refresh(self) -> None: - """Schedule a throttled refresh to avoid glitching.""" - self._needs_refresh = True - if self._refresh_task is None or self._refresh_task.done(): - self._refresh_task = asyncio.create_task(self._throttled_refresh()) - - async def _throttled_refresh(self) -> None: - """Refresh at most every 16ms (~60fps) to avoid glitching.""" - min_interval = 0.016 # ~60fps - while self._needs_refresh: - now = time.monotonic() - elapsed = now - self._last_refresh - if elapsed < min_interval: - await asyncio.sleep(min_interval - elapsed) - self._needs_refresh = False - self._last_refresh = time.monotonic() - self.refresh() - # Small delay to batch rapid updates - await asyncio.sleep(0.008) - - # Regex to filter escape sequences that pyte doesn't handle - # Kitty keyboard protocol, bracketed paste mode queries, etc. - _UNSUPPORTED_ESCAPES = re.compile( - r"\x1b\[[\=\>\<][0-9;]*[a-zA-Z]" # Kitty keyboard protocol (=, >, or < prefix) - r"|\x1b\[\?[0-9;]*u" # Kitty keyboard query - r"|\x1b\[\?[0-9;]*c" # Device attributes query - r"|\x1b\[>[0-9;]*c" # Secondary device attributes - r"|\x1b\]\d+;[^\x07\x1b]*(?:\x07|\x1b\\)" # OSC sequences (title, etc.) - r"|\x1b\[\?2026[hl]" # Synchronized update mode (not used by pyte) - ) - - def _filter_unsupported_escapes(self, data: str) -> str: - """Filter out escape sequences that pyte doesn't handle.""" - return self._UNSUPPORTED_ESCAPES.sub("", data) - - async def _read_output(self) -> None: - """Read output from the PTY.""" - if self._master_fd is None: - return - - while True: - try: - # Wait for data to be available - await asyncio.sleep(0.005) - - try: - data = os.read(self._master_fd, 65536) - if not data: - break - - # Use incremental decoder to handle partial UTF-8 sequences - # at buffer boundaries (prevents box-drawing chars getting corrupted) - text = self._decoder.decode(data) - if not text: - continue # Still waiting for more bytes to complete a sequence - text = self._filter_unsupported_escapes(text) - - # Feed data to pyte - self._stream.feed(text) - self._schedule_refresh() - - except BlockingIOError: - # No data available - continue - except OSError: - # PTY closed - break - - except asyncio.CancelledError: - break - - def stop(self) -> None: - """Stop the terminal process.""" - if self._read_task: - self._read_task.cancel() - self._read_task = None - - if self._refresh_task: - self._refresh_task.cancel() - self._refresh_task = None - - if self._master_fd is not None: - try: - os.close(self._master_fd) - except OSError: - pass - self._master_fd = None - - if self._pid is not None: - try: - os.kill(self._pid, 9) - os.waitpid(self._pid, 0) - except (OSError, ChildProcessError): - pass - self._pid = None - - def is_running(self) -> bool: - """Check if the process is still running.""" - if self._pid is None: - return False - - try: - pid, status = os.waitpid(self._pid, os.WNOHANG) - if pid == 0: - return True # Still running - else: - self._pid = None - return False - except ChildProcessError: - self._pid = None - return False - - def send(self, data: str) -> None: - """Send data to the terminal.""" - if self._master_fd is not None: - try: - os.write(self._master_fd, data.encode("utf-8")) - except OSError: - pass - - # ANSI 256-color palette (standard 16 colors) - ANSI_COLORS = { - "black": "#000000", - "red": "#cd0000", - "green": "#00cd00", - "yellow": "#cdcd00", - "blue": "#0000ee", - "magenta": "#cd00cd", - "cyan": "#00cdcd", - "white": "#e5e5e5", - "brightblack": "#7f7f7f", - "brightred": "#ff0000", - "brightgreen": "#00ff00", - "brightyellow": "#ffff00", - "brightblue": "#5c5cff", - "brightmagenta": "#ff00ff", - "brightcyan": "#00ffff", - "brightwhite": "#ffffff", - } - - def _convert_color(self, color: str, is_bg: bool = False) -> str | None: - """Convert pyte color to Rich color string.""" - if not color or color == "default": - return None - - # Handle hex colors (with or without # prefix) - if color.startswith("#"): - return color - - # Check if it's a hex color without # (pyte returns "ff0000" not "#ff0000") - if len(color) == 6 and all(c in "0123456789abcdefABCDEF" for c in color): - return f"#{color}" - - # Handle named ANSI colors - color_lower = color.lower() - if color_lower in self.ANSI_COLORS: - return self.ANSI_COLORS[color_lower] - - # Handle 256-color palette (numeric) - try: - num = int(color) - if 0 <= num <= 255: - return f"color({num})" - except ValueError: - pass - - # Fallback: try to use the color name directly - return color - - def _get_line_at(self, y: int): - """Get the line at display position y, accounting for scroll offset. - - When scrolled, we show lines from history mixed with current buffer. - scroll_offset=0 means showing current screen. - scroll_offset=N means the top of display shows N lines back in history. - """ - history = self._screen.history.top - history_len = len(history) - - if self._scroll_offset == 0: - # Not scrolled - show current buffer - return self._screen.buffer[y] - - # Calculate which line to show - # Display line 0 should show history[history_len - scroll_offset] - # Display line N should show history[history_len - scroll_offset + N] - # If that index >= history_len, we're into the current buffer - - history_index = history_len - self._scroll_offset + y - - if history_index < 0: - # Before start of history - return blank - return None - elif history_index < history_len: - # In history - return history[history_index] - else: - # In current buffer - buffer_index = history_index - history_len - if buffer_index < self._rows: - return self._screen.buffer[buffer_index] - return None - - def render_line(self, y: int) -> Strip: - """Render a line of the terminal.""" - # Screen buffer dimensions (what pyte has) - screen_cols = self._cols - screen_rows = self._rows - - # Widget display dimensions (what we need to output) - output_width = ( - self.size.width - if self.size.width > 0 - else screen_cols + self.PADDING_LEFT + self.PADDING_RIGHT - ) - - if y >= screen_rows: - return Strip.blank(output_width) - - # Get the line to render, accounting for scroll offset - line = self._get_line_at(y) - if line is None: - return Strip.blank(output_width) - - text = Text() - - # Add left padding (uses terminal background) - text.append(" " * self.PADDING_LEFT) - - # Only access indices within the screen buffer - cols_to_render = min(screen_cols, output_width - self.PADDING_LEFT - self.PADDING_RIGHT) - - # Debug logging for render pipeline - debug_logger = pyte.get_debug_logger() - log_this_line = False - - for x in range(cols_to_render): - char = line[x] - char_data = char.data if char.data else " " - - # Check for box-drawing and other potentially problematic characters - if len(char_data) == 1: - code = ord(char_data) - # Log box-drawing characters (U+2500-U+257F) - if 0x2500 <= code <= 0x257F: - log_this_line = True - if debug_logger: - debug_logger( - f"RENDER y={y} x={x}: box-drawing U+{code:04X} char='{char_data}'" - ) - - # Handle characters that may not render correctly - if len(char_data) == 1: - code = ord(char_data) - # Control characters (except space) - if code < 32 and code != 0 or 127 <= code <= 159: - char_data = " " - # Braille patterns (U+2800-U+28FF) - used for spinners - # Replace with simple ASCII spinner chars or spaces - elif 0x2800 <= code <= 0x28FF: - # Map braille spinner to simple dots - char_data = "·" - - # Build style from pyte character attributes - style_parts = [] - - # Foreground color - fg = self._convert_color(char.fg) - if fg: - style_parts.append(fg) - - # Background color - bg = self._convert_color(char.bg, is_bg=True) - if bg: - style_parts.append(f"on {bg}") - - # Text attributes - if char.bold: - style_parts.append("bold") - if char.italics: - style_parts.append("italic") - if char.underscore: - style_parts.append("underline") - if char.strikethrough: - style_parts.append("strike") - if char.reverse: - style_parts.append("reverse") - - style = " ".join(style_parts) if style_parts else None - text.append(char_data, style=style) - - # Pad with spaces to fill remaining width (includes right padding) - content_width = self.PADDING_LEFT + cols_to_render - if output_width > content_width: - text.append(" " * (output_width - content_width)) - - # Debug: log the Rich Text content before rendering - if log_this_line and debug_logger: - debug_logger(f"RENDER y={y}: Rich Text plain='{text.plain[:80]}...'") - - # Render text to segments for Strip - segments = list(text.render(self.app.console)) - - # Debug: log segments if we had box-drawing chars - if log_this_line and debug_logger: - for i, seg in enumerate(segments[:10]): # First 10 segments - seg_text = seg.text if hasattr(seg, "text") else str(seg) - if len(seg_text) <= 5: - debug_logger(f"RENDER y={y} seg[{i}]: '{seg_text}' (repr: {repr(seg_text)})") - - strip = Strip(segments) - - # Ensure strip is exactly the output width - if strip.cell_length != output_width: - strip = strip.crop_extend(0, output_width, None) - - return strip - - def on_key(self, event) -> None: - """Handle key presses.""" - # Map special keys - key_map = { - "enter": "\r", - "tab": "\t", - "backspace": "\x7f", - "delete": "\x1b[3~", - "up": "\x1b[A", - "down": "\x1b[B", - "right": "\x1b[C", - "left": "\x1b[D", - "home": "\x1b[H", - "end": "\x1b[F", - "pageup": "\x1b[5~", - "pagedown": "\x1b[6~", - "escape": "\x1b", - } - - if event.key in key_map: - self.send(key_map[event.key]) - event.prevent_default() - event.stop() - elif event.key == "ctrl+c": - self.send("\x03") - event.prevent_default() - event.stop() - elif event.key == "ctrl+d": - self.send("\x04") - event.prevent_default() - event.stop() - elif event.key == "ctrl+z": - self.send("\x1a") - event.prevent_default() - event.stop() - elif event.key == "ctrl+l": - self.send("\x0c") - event.prevent_default() - event.stop() - elif event.key == "shift+pageup": - # Scroll up in history - self._scroll_up(self._rows // 2) - event.prevent_default() - event.stop() - elif event.key == "shift+pagedown": - # Scroll down in history - self._scroll_down(self._rows // 2) - event.prevent_default() - event.stop() - elif event.key == "shift+home": - # Scroll to top of history - self._scroll_to_top() - event.prevent_default() - event.stop() - elif event.key == "shift+end": - # Scroll to bottom (current) - self._scroll_to_bottom() - event.prevent_default() - event.stop() - elif event.character and len(event.character) == 1: - # Any typing scrolls to bottom - self._scroll_to_bottom() - self.send(event.character) - event.prevent_default() - event.stop() - - def on_mouse_scroll_up(self, event) -> None: - """Handle mouse scroll up (view older content).""" - self._scroll_up(3) - event.prevent_default() - event.stop() - - def on_mouse_scroll_down(self, event) -> None: - """Handle mouse scroll down (view newer content).""" - self._scroll_down(3) - event.prevent_default() - event.stop() - - def _scroll_up(self, lines: int) -> None: - """Scroll up (back in history) by given number of lines.""" - max_scroll = len(self._screen.history.top) - new_offset = min(self._scroll_offset + lines, max_scroll) - if new_offset != self._scroll_offset: - self._scroll_offset = new_offset - self.refresh() - - def _scroll_down(self, lines: int) -> None: - """Scroll down (forward toward current) by given number of lines.""" - new_offset = max(self._scroll_offset - lines, 0) - if new_offset != self._scroll_offset: - self._scroll_offset = new_offset - self.refresh() - - def _scroll_to_top(self) -> None: - """Scroll to the top of history.""" - max_scroll = len(self._screen.history.top) - if self._scroll_offset != max_scroll: - self._scroll_offset = max_scroll - self.refresh() - - def _scroll_to_bottom(self) -> None: - """Scroll to the bottom (current screen).""" - if self._scroll_offset != 0: - self._scroll_offset = 0 - self.refresh() - - @property - def history_size(self) -> int: - """Get the number of lines in scrollback history.""" - return len(self._screen.history.top) - - class ClaudePanel(Vertical): """Terminal panel running Claude Code CLI. diff --git a/clide/widgets/panels/workspace.py b/clide/widgets/panels/workspace.py index 281d2b05..8aa3f8a5 100644 --- a/clide/widgets/panels/workspace.py +++ b/clide/widgets/panels/workspace.py @@ -95,13 +95,6 @@ class WorkspacePanel(Vertical): self.file_path = file_path super().__init__() - class CommandSubmitted(Message): - """Emitted when a terminal command is submitted.""" - - def __init__(self, command: str) -> None: - self.command = command - super().__init__() - class CloseRequested(Message): """Emitted when workspace should be hidden.""" @@ -401,32 +394,7 @@ class WorkspacePanel(Vertical): self.show("terminal") try: terminal = self.query_one("#terminal-pane", TerminalPane) - terminal_input = terminal.query_one("#terminal-input") - terminal_input.focus() - except Exception: - pass - - def write_terminal_output(self, text: str, style: str = "output") -> None: - """Write output to terminal.""" - try: - terminal = self.query_one("#terminal-pane", TerminalPane) - terminal.write_output(text, style) - except Exception: - pass - - def write_terminal_error(self, error: str) -> None: - """Write error to terminal.""" - try: - terminal = self.query_one("#terminal-pane", TerminalPane) - terminal.write_error(error) - except Exception: - pass - - def clear_terminal(self) -> None: - """Clear terminal output.""" - try: - terminal = self.query_one("#terminal-pane", TerminalPane) - terminal.clear() + terminal.focus_terminal() except Exception: pass @@ -442,10 +410,3 @@ class WorkspacePanel(Vertical): def on_diff_pane_reject_clicked(self, event: DiffPane.RejectClicked) -> None: """Forward diff reject event.""" self.post_message(self.DiffRejected(event.file_path)) - - def on_terminal_pane_command_submitted( - self, - event: TerminalPane.CommandSubmitted, - ) -> None: - """Forward terminal command event.""" - self.post_message(self.CommandSubmitted(event.command))