Add mouse wheel and keyboard scrolling support to REPL

- Add scroll_offset to REPLState (max 50 lines)
- Modify _render_repl() to use manual scroll position
- Add scroll_output(delta) method for scroll control
- Add PageUp/PageDown keyboard support (scroll 10 lines)
- Add mouse wheel support via SGR mouse tracking
- Update HUD to show scroll percentage (like vim) and position
- Reset scroll when new output arrives
- Add tests for scroll functionality
This commit is contained in:
2026-03-22 17:26:14 -07:00
parent e5799a346a
commit bfcad4963a
6 changed files with 257 additions and 21 deletions

View File

@@ -108,6 +108,22 @@ Pipeline effects:
Effect 'repl' set to off Effect 'repl' set to off
``` ```
## Scrolling Support
The REPL output buffer supports scrolling through command history:
**Keyboard Controls:**
- **PageUp** - Scroll up 10 lines
- **PageDown** - Scroll down 10 lines
- **Mouse wheel up** - Scroll up 3 lines
- **Mouse wheel down** - Scroll down 3 lines
**Scroll Features:**
- **Scroll percentage** shown in HUD (like vim, e.g., "50%")
- **Scroll position** shown in output line (e.g., "(5/20)")
- **Auto-reset** - Scroll resets to bottom when new output arrives
- **Max buffer** - 50 lines (excluding empty lines)
## Notes ## Notes
- The REPL effect needs a content source to overlay on (e.g., headlines, poetry, empty) - The REPL effect needs a content source to overlay on (e.g., headlines, poetry, empty)

View File

@@ -575,8 +575,21 @@ def run_pipeline_mode_direct():
repl_effect.navigate_history(-1) repl_effect.navigate_history(-1)
elif key == "down": elif key == "down":
repl_effect.navigate_history(1) repl_effect.navigate_history(1)
elif key == "page_up":
repl_effect.scroll_output(-10) # Scroll up 10 lines
elif key == "page_down":
repl_effect.scroll_output(10) # Scroll down 10 lines
elif key == "backspace": elif key == "backspace":
repl_effect.backspace() repl_effect.backspace()
elif key.startswith("mouse:"):
# Mouse event format: mouse:button:x:y
parts = key.split(":")
if len(parts) >= 2:
button = int(parts[1])
if button == 64: # Wheel up
repl_effect.scroll_output(-3) # Scroll up 3 lines
elif button == 65: # Wheel down
repl_effect.scroll_output(3) # Scroll down 3 lines
elif len(key) == 1: elif len(key) == 1:
repl_effect.append_to_command(key) repl_effect.append_to_command(key)
# --- End REPL Input Handling --- # --- End REPL Input Handling ---

View File

@@ -1016,8 +1016,21 @@ def run_pipeline_mode(preset_name: str = "demo", graph_config: str | None = None
repl_effect.navigate_history(-1) repl_effect.navigate_history(-1)
elif key == "down": elif key == "down":
repl_effect.navigate_history(1) repl_effect.navigate_history(1)
elif key == "page_up":
repl_effect.scroll_output(-10) # Scroll up 10 lines
elif key == "page_down":
repl_effect.scroll_output(10) # Scroll down 10 lines
elif key == "backspace": elif key == "backspace":
repl_effect.backspace() repl_effect.backspace()
elif key.startswith("mouse:"):
# Mouse event format: mouse:button:x:y
parts = key.split(":")
if len(parts) >= 2:
button = int(parts[1])
if button == 64: # Wheel up
repl_effect.scroll_output(-3) # Scroll up 3 lines
elif button == 65: # Wheel down
repl_effect.scroll_output(3) # Scroll down 3 lines
elif len(key) == 1: elif len(key) == 1:
repl_effect.append_to_command(key) repl_effect.append_to_command(key)
# --- End REPL Input Handling --- # --- End REPL Input Handling ---

View File

@@ -157,6 +157,9 @@ class TerminalDisplay:
def cleanup(self) -> None: def cleanup(self) -> None:
from engine.terminal import CURSOR_ON from engine.terminal import CURSOR_ON
# Disable mouse tracking if enabled
self.disable_mouse_tracking()
# Restore normal terminal mode if raw mode was enabled # Restore normal terminal mode if raw mode was enabled
self.set_raw_mode(False) self.set_raw_mode(False)
@@ -174,6 +177,24 @@ class TerminalDisplay:
"""Request quit (e.g., when Ctrl+C is pressed).""" """Request quit (e.g., when Ctrl+C is pressed)."""
self._quit_requested = True self._quit_requested = True
def enable_mouse_tracking(self) -> None:
"""Enable SGR mouse tracking mode."""
try:
# SGR mouse mode: \x1b[?1006h
sys.stdout.write("\x1b[?1006h")
sys.stdout.flush()
except (OSError, AttributeError):
pass # Terminal might not support mouse tracking
def disable_mouse_tracking(self) -> None:
"""Disable SGR mouse tracking mode."""
try:
# Disable SGR mouse mode: \x1b[?1006l
sys.stdout.write("\x1b[?1006l")
sys.stdout.flush()
except (OSError, AttributeError):
pass
def set_raw_mode(self, enable: bool = True) -> None: def set_raw_mode(self, enable: bool = True) -> None:
"""Enable/disable raw terminal mode for input capture. """Enable/disable raw terminal mode for input capture.
@@ -192,7 +213,11 @@ class TerminalDisplay:
# Set raw mode # Set raw mode
tty.setraw(sys.stdin.fileno()) tty.setraw(sys.stdin.fileno())
self._raw_mode_enabled = True self._raw_mode_enabled = True
# Enable mouse tracking
self.enable_mouse_tracking()
elif not enable and self._raw_mode_enabled: elif not enable and self._raw_mode_enabled:
# Disable mouse tracking
self.disable_mouse_tracking()
# Restore original terminal settings # Restore original terminal settings
if self._original_termios: if self._original_termios:
termios.tcsetattr( termios.tcsetattr(
@@ -223,19 +248,38 @@ class TerminalDisplay:
char = sys.stdin.read(1) char = sys.stdin.read(1)
if char == "\x1b": # Escape sequence if char == "\x1b": # Escape sequence
# Read next character to determine key # Read next characters to determine key
seq = sys.stdin.read(2) # Try to read up to 10 chars for longer sequences
if seq == "[A": seq = sys.stdin.read(10)
# PageUp: \x1b[5~
if seq.startswith("[5~"):
keys.append("page_up")
# PageDown: \x1b[6~
elif seq.startswith("[6~"):
keys.append("page_down")
# Arrow keys: \x1b[A, \x1b[B, etc.
elif seq.startswith("["):
if seq[1] == "A":
keys.append("up") keys.append("up")
elif seq == "[B": elif seq[1] == "B":
keys.append("down") keys.append("down")
elif seq == "[C": elif seq[1] == "C":
keys.append("right") keys.append("right")
elif seq == "[D": elif seq[1] == "D":
keys.append("left") keys.append("left")
else: else:
# Unknown escape sequence # Unknown escape sequence
keys.append("escape") keys.append("escape")
# Mouse events: \x1b[<B;X;Ym or \x1b[<B;X;YM
elif seq.startswith("[<"):
mouse_seq = "\x1b" + seq
mouse_data = self._parse_mouse_event(mouse_seq)
if mouse_data:
keys.append(mouse_data)
else:
# Unknown escape sequence
keys.append("escape")
elif char == "\n" or char == "\r": elif char == "\n" or char == "\r":
keys.append("return") keys.append("return")
elif char == "\t": elif char == "\t":
@@ -248,8 +292,6 @@ class TerminalDisplay:
keys.append("ctrl_c") keys.append("ctrl_c")
elif char == "\x04": # Ctrl+D elif char == "\x04": # Ctrl+D
keys.append("ctrl_d") keys.append("ctrl_d")
elif char == "\x1b": # Escape
keys.append("escape")
elif char.isprintable(): elif char.isprintable():
keys.append(char) keys.append(char)
except OSError: except OSError:
@@ -257,6 +299,40 @@ class TerminalDisplay:
return keys return keys
def _parse_mouse_event(self, data: str) -> str | None:
"""Parse SGR mouse event sequence.
Format: \x1b[<B;X;Ym (release) or \x1b[<B;X;YM (press)
B = button number (0=left, 1=middle, 2=right, 64=wheel up, 65=wheel down)
X, Y = coordinates (1-indexed)
Returns:
Mouse event string like "mouse:64:10:5" or None if not a mouse event
"""
if not data.startswith("\x1b[<"):
return None
# Find the ending 'm' or 'M'
end_pos = data.rfind("m")
if end_pos == -1:
end_pos = data.rfind("M")
if end_pos == -1:
return None
inner = data[3:end_pos] # Remove \x1b[< and trailing m/M
parts = inner.split(";")
if len(parts) >= 3:
try:
button = int(parts[0])
x = int(parts[1]) - 1 # Convert to 0-indexed
y = int(parts[2]) - 1
return f"mouse:{button}:{x}:{y}"
except ValueError:
pass
return None
def is_raw_mode_enabled(self) -> bool: def is_raw_mode_enabled(self) -> bool:
"""Check if raw mode is currently enabled.""" """Check if raw mode is currently enabled."""
return self._raw_mode_enabled return self._raw_mode_enabled

View File

@@ -47,8 +47,9 @@ class REPLState:
current_command: str = "" current_command: str = ""
history_index: int = -1 history_index: int = -1
output_buffer: list[str] = field(default_factory=list) output_buffer: list[str] = field(default_factory=list)
scroll_offset: int = 0 # Manual scroll position (0 = bottom of buffer)
max_history: int = 50 max_history: int = 50
max_output_lines: int = 20 max_output_lines: int = 50 # 50 lines excluding empty lines
class ReplEffect(EffectPlugin): class ReplEffect(EffectPlugin):
@@ -137,10 +138,23 @@ class ReplEffect(EffectPlugin):
# Line 1: Title + FPS + Frame time # Line 1: Title + FPS + Frame time
fps_str = f"FPS: {fps:.1f}" if fps > 0 else "FPS: --" fps_str = f"FPS: {fps:.1f}" if fps > 0 else "FPS: --"
time_str = f"{frame_time:.1f}ms" if frame_time > 0 else "--ms" time_str = f"{frame_time:.1f}ms" if frame_time > 0 else "--ms"
# Calculate scroll percentage (like vim)
scroll_pct = 0
if len(self.state.output_buffer) > 1:
max_scroll = len(self.state.output_buffer) - 1
scroll_pct = (
int((self.state.scroll_offset / max_scroll) * 100)
if max_scroll > 0
else 0
)
scroll_str = f"{scroll_pct}%"
line1 = ( line1 = (
f"\033[38;5;46mMAINLINE REPL\033[0m " f"\033[38;5;46mMAINLINE REPL\033[0m "
f"\033[38;5;245m|\033[0m \033[38;5;39m{fps_str}\033[0m " f"\033[38;5;245m|\033[0m \033[38;5;39m{fps_str}\033[0m "
f"\033[38;5;245m|\033[0m \033[38;5;208m{time_str}\033[0m" f"\033[38;5;245m|\033[0m \033[38;5;208m{time_str}\033[0m "
f"\033[38;5;245m|\033[0m \033[38;5;220m{scroll_str}\033[0m"
) )
lines.append(line1[:width]) lines.append(line1[:width])
@@ -156,9 +170,14 @@ class ReplEffect(EffectPlugin):
) )
lines.append(line2[:width]) lines.append(line2[:width])
# Line 3: Output buffer count # Line 3: Output buffer count with scroll indicator
out_count = len(self.state.output_buffer) out_count = len(self.state.output_buffer)
line3 = f"\033[38;5;44mOUTPUT:\033[0m \033[1;38;5;227m{out_count}\033[0m lines" scroll_pos = f"({self.state.scroll_offset}/{out_count})"
line3 = (
f"\033[38;5;44mOUTPUT:\033[0m "
f"\033[1;38;5;227m{out_count}\033[0m lines "
f"\033[38;5;245m{scroll_pos}\033[0m"
)
lines.append(line3[:width]) lines.append(line3[:width])
return lines return lines
@@ -170,12 +189,16 @@ class ReplEffect(EffectPlugin):
# Calculate how many output lines to show # Calculate how many output lines to show
# Reserve 1 line for input prompt # Reserve 1 line for input prompt
output_height = height - 1 output_height = height - 1
output_start = max(0, len(self.state.output_buffer) - output_height)
# Manual scroll: scroll_offset=0 means show bottom of buffer
# scroll_offset increases as you scroll up through history
buffer_len = len(self.state.output_buffer)
output_start = max(0, buffer_len - output_height - self.state.scroll_offset)
# Render output buffer # Render output buffer
for i in range(output_height): for i in range(output_height):
idx = output_start + i idx = output_start + i
if idx < len(self.state.output_buffer): if idx < buffer_len:
line = self.state.output_buffer[idx][:width] line = self.state.output_buffer[idx][:width]
lines.append(line) lines.append(line)
else: else:
@@ -191,6 +214,25 @@ class ReplEffect(EffectPlugin):
return lines return lines
def scroll_output(self, delta: int) -> None:
"""Scroll the output buffer by delta lines.
Args:
delta: Positive to scroll up (back in time), negative to scroll down
"""
if not self.state.output_buffer:
return
# Calculate max scroll (can't scroll past top of buffer)
max_scroll = max(0, len(self.state.output_buffer) - 1)
# Update scroll offset
self.state.scroll_offset = max(
0, min(max_scroll, self.state.scroll_offset + delta)
)
# Reset scroll when new output arrives (handled in process_command)
def _get_metrics(self, ctx: EffectContext) -> dict: def _get_metrics(self, ctx: EffectContext) -> dict:
"""Get pipeline metrics from context.""" """Get pipeline metrics from context."""
metrics = ctx.get_state("metrics") metrics = ctx.get_state("metrics")
@@ -230,6 +272,9 @@ class ReplEffect(EffectPlugin):
# Add to output buffer # Add to output buffer
self.state.output_buffer.append(f"> {cmd}") self.state.output_buffer.append(f"> {cmd}")
# Reset scroll offset when new output arrives (scroll to bottom)
self.state.scroll_offset = 0
# Parse command # Parse command
parts = cmd.split() parts = cmd.split()
cmd_name = parts[0].lower() cmd_name = parts[0].lower()
@@ -322,13 +367,17 @@ class ReplEffect(EffectPlugin):
self.state.output_buffer.append("No context available") self.state.output_buffer.append("No context available")
def _cmd_available(self, ctx: EffectContext | None): def _cmd_available(self, ctx: EffectContext | None):
"""List all available effect types.""" """List all available effect types and stage categories."""
try: try:
from engine.effects import get_registry from engine.effects import get_registry
from engine.effects.plugins import discover_plugins from engine.effects.plugins import discover_plugins
from engine.pipeline.registry import StageRegistry, discover_stages
# Discover plugins if not already done # Discover plugins and stages if not already done
discover_plugins() discover_plugins()
discover_stages()
# List effect types from registry
registry = get_registry() registry = get_registry()
all_effects = registry.list_all() all_effects = registry.list_all()
@@ -338,8 +387,20 @@ class ReplEffect(EffectPlugin):
self.state.output_buffer.append(f" - {name}") self.state.output_buffer.append(f" - {name}")
else: else:
self.state.output_buffer.append("No effects registered") self.state.output_buffer.append("No effects registered")
# List stage categories and their types
categories = StageRegistry.list_categories()
if categories:
self.state.output_buffer.append("")
self.state.output_buffer.append("Stage categories:")
for category in sorted(categories):
stages = StageRegistry.list(category)
if stages:
self.state.output_buffer.append(f" {category}:")
for stage_name in sorted(stages):
self.state.output_buffer.append(f" - {stage_name}")
except Exception as e: except Exception as e:
self.state.output_buffer.append(f"Error listing effects: {e}") self.state.output_buffer.append(f"Error listing available types: {e}")
def _cmd_effect(self, args: list[str], ctx: EffectContext | None): def _cmd_effect(self, args: list[str], ctx: EffectContext | None):
"""Toggle effect on/off.""" """Toggle effect on/off."""

View File

@@ -199,3 +199,60 @@ class TestReplConfig:
assert repl.config.enabled is False assert repl.config.enabled is False
assert repl.config.intensity == 0.5 assert repl.config.intensity == 0.5
assert repl.config.params["display_height"] == 10 assert repl.config.params["display_height"] == 10
class TestReplScrolling:
"""Tests for REPL scrolling functionality."""
@pytest.fixture(autouse=True)
def setup(self):
"""Setup before each test."""
self.repl = ReplEffect()
def test_scroll_offset_initial(self):
"""Scroll offset starts at 0."""
assert self.repl.state.scroll_offset == 0
def test_scroll_output_positive(self):
"""Scrolling with positive delta moves back through buffer."""
# Add some output
self.repl.state.output_buffer = [f"line{i}" for i in range(20)]
# Scroll up 5 lines
self.repl.scroll_output(5)
assert self.repl.state.scroll_offset == 5
def test_scroll_output_negative(self):
"""Scrolling with negative delta moves forward through buffer."""
# Add some output and scroll up first
self.repl.state.output_buffer = [f"line{i}" for i in range(20)]
self.repl.state.scroll_offset = 10
# Scroll down 3 lines
self.repl.scroll_output(-3)
assert self.repl.state.scroll_offset == 7
def test_scroll_output_bounds(self):
"""Scroll offset stays within valid bounds."""
# Add some output
self.repl.state.output_buffer = [f"line{i}" for i in range(10)]
# Try to scroll past top
self.repl.scroll_output(100)
assert self.repl.state.scroll_offset == 9 # max: len(output) - 1
# Try to scroll past bottom
self.repl.state.scroll_offset = 5
self.repl.scroll_output(-100)
assert self.repl.state.scroll_offset == 0
def test_scroll_resets_on_new_output(self):
"""Scroll offset resets when new command output arrives."""
self.repl.state.output_buffer = [f"line{i}" for i in range(20)]
self.repl.state.scroll_offset = 10
# Process a new command
self.repl.process_command("test command")
# Scroll offset should be reset to 0
assert self.repl.state.scroll_offset == 0