fix: scroll_history buffer manuale per pyte

- Teniamo il nostro history list manualmente
- screen.display per contenuto corrente
- Nessun accesso a screen.lines o screen.scroll_history come list
This commit is contained in:
opencode committed 2026-03-24 17:57:41 +01:00
1 parent b6e8864829
commit e69a289361
1 file changed
+90 -125
+90 -125
View File
@@ -1,6 +1,6 @@
"""
Terminal UI con scrollback usando Pyte
Versione con navigazione scrollback
Versione con buffer manuale per scrollback
"""
import os
import sys
@@ -20,7 +20,7 @@ class TerminalUI:
Terminal emulator con scrollback usando pyte.
Features:
- Scrollback buffer completo
- Scrollback buffer manuale
- Navigazione con frecce e PgUp/PgDn
- Output colorato
"""
@@ -43,11 +43,11 @@ class TerminalUI:
# Crea screen pyte
self.screen = pyte.Screen(self.cols, self.rows)
self.screen.scroll_history = scrollback
# Stream per feed input
self.stream = pyte.Stream(self.screen)
# Il nostro buffer per lo scrollback
self.history: list = []
# Scroll offset (0 = latest)
self.scroll_offset = 0
@@ -63,103 +63,82 @@ class TerminalUI:
# Callback agent
self.agent_callback: Optional[Callable[[str], str]] = None
# Signal handler
self._old_sigint: Optional[signal.Handler] = None
def _add_to_history(self, line: str):
"""Aggiunge una linea allo history buffer"""
self.history.append(line)
if len(self.history) > self.scrollback:
self.history.pop(0)
def _draw(self):
"""Ridisegna lo schermo"""
reset = "\033[0m"
# Clear
sys.stdout.write("\033[2J\033[H")
# Header con colori
# Header
header_color = "\033[1;36m"
reset = "\033[0m"
header_line = "═" * (self.cols - 4)
sys.stdout.write(f"{header_color}╔{header_line}╗{reset}\n")
sys.stdout.write(f"{header_color}╔{'═' * (self.cols - 4)}╗{reset}\n")
title = " LangCode Agent - pyte UI "
title_pos = (self.cols - len(title)) // 2
sys.stdout.write(f"{header_color}║{reset}" + " " * title_pos + f"\033[1;33m{title}\033[0m" + " " * (self.cols - title_pos - len(title) - 3) + f"{header_color}║{reset}\n")
sys.stdout.write(f"{header_color}╚{header_line}╝{reset}\n")
sys.stdout.write(f"{header_color}╚{'═' * (self.cols - 4)}║{reset}\n")
# Contenuto scrollback
# Lo screen.display contiene le linee visibili
# Ma dobbiamo mostrare anche lo scroll_history
if self.mode == "scroll" or self.scroll_offset > 0:
# Mostra modalità scroll
visible_content = self._get_visible_content()
for i, line in enumerate(visible_content):
# Tronca o pad la linea
display = line[:self.cols - 2] if len(line) > self.cols - 2 else line
sys.stdout.write(display + " " * (self.cols - len(display) - 1) + "\n")
else:
# Modalità input normale - mostra output recente
# Lo screen.display ha le ultime self.rows linee
for line in self.screen.display:
# Rimuovi escape codes per display puro
clean = self._strip_ansi(line.rstrip())
if clean: # Solo linee non vuote
sys.stdout.write(clean + "\n")
# Status bar - scroll_history è un int, lines contiene il contenuto
status_color = "\033[1;34m"
total_lines = len(self.screen.lines) + (self.scrollback if hasattr(self, 'scrollback') else 10000)
scroll_info = f" Linee buffer: ~{total_lines} | Modalità: {self.mode} "
# Contenuto
if self.mode == "scroll":
scroll_info += f" | Offset: {self.scroll_offset} | q=exit scroll | ENTER=exit "
# Mostra history buffer
visible = self.rows - 8
total = len(self.history)
start = max(0, total - self.scroll_offset - visible)
end = min(total, start + visible)
for i in range(start, end):
line = self.history[i][:self.cols - 2]
sys.stdout.write(line + " " * (self.cols - len(line) - 1) + "\n")
else:
# Modalità input - mostra screen.display
for line in self.screen.display:
clean = self._strip_ansi(line.rstrip())
if clean:
display = clean[:self.cols - 2]
sys.stdout.write(display + " " * (self.cols - len(display) - 1) + "\n")
# Status bar
status_color = "\033[1;34m"
total_lines = len(self.history)
scroll_info = f" History: {total_lines} linee | Modalità: {self.mode} "
if self.mode == "scroll":
scroll_info += f" | Offset: {self.scroll_offset} | q/ENTER=exit scroll "
else:
scroll_info += f" | PgUp=scroll | q=quit "
sys.stdout.write(f"\n{status_color}{'─' * (self.cols - len(scroll_info) - 1)}{reset}{scroll_info}{reset}\n")
# Input line o prompt
status_bar_width = self.cols - len(scroll_info) - 1
sys.stdout.write(f"\n{status_color}{'─' * status_bar_width}{reset}{scroll_info}{reset}\n")
# Input line
if self.mode == "input":
sys.stdout.write(f"\033[1;35m{self.prompt}\033[0m {self.input_buffer}")
# Move cursor back if needed
if self.cursor_pos < len(self.input_buffer):
move_back = len(self.input_buffer) - self.cursor_pos
sys.stdout.write(f"\033[{move_back}D")
else:
sys.stdout.write(f"\033[1;33m[SCROLL MODE] Premi q o ENTER per uscire dallo scroll\033[0m")
sys.stdout.write(f"\033[1;33m[SCROLL MODE] q o ENTER per uscire\033[0m")
sys.stdout.flush()
def _get_visible_content(self) -> list:
"""Ottieni contenuto visibile considerando lo scroll"""
# In pyte: screen.lines contiene le linee correnti
# screen.display è la rappresentazione visibile corrente
# Non c'è uno scroll_history come lista - dobbiamo tenere il nostro buffer
# Usa il contenuto corrente dello screen
all_lines = list(self.screen.display)
if not all_lines:
return []
# Calcola offset dal fondo
total = len(all_lines)
visible = self.rows - 8 # -header, status, prompt, etc.
start = max(0, total - self.scroll_offset - visible)
end = total - self.scroll_offset
if end <= start:
end = start + visible
return all_lines[start:end]
def _strip_ansi(self, text: str) -> str:
"""Rimuovi codici ANSI da una stringa"""
"""Rimuovi codici ANSI"""
import re
ansi_pattern = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
return ansi_pattern.sub('', text)
def _handle_key(self, char: str):
"""Gestisce un singolo carattere input"""
"""Gestisce un singolo carattere"""
if char == '\x1b': # ESC
self.mode = "scroll"
self.scroll_offset = len(self.screen.scroll_history) + len(self.screen.display) - self.rows + 5
self.scroll_offset = len(self.history)
return
if self.mode == "scroll":
@@ -169,17 +148,14 @@ class TerminalUI:
elif char == '\r' or char == '\n':
self.mode = "input"
self.scroll_offset = 0
elif char in ['\x7f', '\x08']: # Backspace in scroll mode
elif char in ['j', 'J']:
self.scroll_offset = min(len(self.history), self.scroll_offset + 1)
elif char in ['k', 'K']:
self.scroll_offset = max(0, self.scroll_offset - 1)
elif char == 'j': # vim-style down
self.scroll_offset = max(0, self.scroll_offset - 1)
elif char == 'k': # vim-style up
self.scroll_offset += 1
return
# Modalità input
if char == '\r' or char == '\n':
# Processa comando
cmd = self.input_buffer
self.input_buffer = ""
self.cursor_pos = 0
@@ -192,13 +168,12 @@ class TerminalUI:
self.stream.feed('\b \b')
elif char == '\x03': # Ctrl+C
return None
elif char == '\x0c': # Ctrl+L (clear)
elif char == '\x0c': # Ctrl+L
self.screen.reset()
self.input_buffer = ""
self.cursor_pos = 0
else:
# Carattere normale
if len(char) == 1 and ord(char) >= 32 and ord(char) <= 126:
if len(char) == 1 and 32 <= ord(char) <= 126:
self.input_buffer = self.input_buffer[:self.cursor_pos] + char + self.input_buffer[self.cursor_pos:]
self.cursor_pos += 1
self.stream.feed(char)
@@ -210,50 +185,47 @@ class TerminalUI:
if not command.strip():
return
# Output comando utente
prompt_color = "\033[1;36m"
reset = "\033[0m"
self.stream.feed(f'\r\n{prompt_color}👤 {command}{reset}\r\n\n')
# Aggiungi al nostro history
self._add_to_history(f"\033[1;36m👤 {command}\033[0m")
# Thinking indicator
thinking_color = "\033[1;33m"
self.stream.feed(f'{thinking_color}🤔 Pensando...{reset}\r\n')
# Feed a pyte
self.stream.feed(f'\r\n\033[1;36m👤 {command}\033[0m\r\n\n')
# Thinking
self._add_to_history(f"\033[1;33m🤔 Pensando...\033[0m")
self.stream.feed(f'\033[1;33m🤔 Pensando...\033[0m\r\n')
# Chiamata agent
try:
response = self.agent_callback(command)
# Rimuovi thinking e mostra risposta
# Scroll back di una linea
if len(self.screen.scroll_history) > 0:
# Il pensiero è nell'ultima linea visibile, rimuoviamolo
pass
if self.history and "🤔 Pensando..." in self.history[-1]:
self.history.pop()
# Output risposta
for line in response.split('\n'):
self._add_to_history(f'\033[1;32m🤖 {line}\033[0m')
self.stream.feed(f'\r\n\033[1;32m🤖 {line}\033[0m')
self.stream.feed('\r\n')
except Exception as e:
self._add_to_history(f'\033[1;31m❌ Errore: {str(e)}\033[0m')
self.stream.feed(f'\r\n\033[1;31m❌ Errore: {str(e)}\033[0m\r\n')
def start(self, agent_callback: Callable[[str], str]):
"""Avvia il terminale interattivo"""
"""Avvia il terminale"""
self.agent_callback = agent_callback
self.running = True
# Signal handler
def handle_sigint(signum, frame):
sys.stdout.write("\n\033[1;33m👋 Arrivederci!\033[0m\n")
sys.stdout.flush()
self.running = False
sys.exit(0)
self._old_sigint = signal.signal(signal.SIGINT, handle_sigint)
signal.signal(signal.SIGINT, handle_sigint)
try:
# Abilita raw mode per catturare singoli tasti
import tty
import termios
@@ -261,14 +233,10 @@ class TerminalUI:
old_settings = termios.tcgetattr(fd)
try:
# Set raw mode
tty.setraw(fd)
# Mostra schermo iniziale
self._draw()
while self.running:
# Leggi un carattere
ch = os.read(fd, 1)
if not ch:
@@ -276,15 +244,14 @@ class TerminalUI:
char = ch.decode('utf-8', errors='replace')
# Gestisci PgUp/PgDn in raw mode
if char == '\x1b': # ESC sequence start
# Leggi il resto della sequenza
# Check for ESC sequence (arrow keys, PgUp, PgDn)
if char == '\x1b':
import select
if select.select([fd], [], [], 0.1)[0]:
seq = os.read(fd, 2).decode('utf-8', errors='replace')
if seq == '[5~': # PgUp
self.mode = "scroll"
self.scroll_offset = max(0, len(self.screen.display) - self.rows + 5)
self.scroll_offset = len(self.history)
self._draw()
continue
elif seq == '[6~': # PgDn
@@ -294,18 +261,17 @@ class TerminalUI:
self.mode = "input"
self._draw()
continue
elif seq == '[A': # Up arrow
elif seq == '[A': # Up
if self.mode == "scroll":
self.scroll_offset += 1
self.scroll_offset = min(len(self.history), self.scroll_offset + 1)
self._draw()
continue
elif seq == '[B': # Down arrow
elif seq == '[B': # Down
if self.mode == "scroll":
self.scroll_offset = max(0, self.scroll_offset - 1)
self._draw()
continue
# Processa il carattere
result = self._handle_key(char)
if result is None:
@@ -316,30 +282,28 @@ class TerminalUI:
self._draw()
continue
if result is None and char == '\x03': # Ctrl+C
break
# Esegui comando
# Process command
if result:
if result.lower() == '/exit':
break
elif result.lower() == '/help':
self.stream.feed('\r\n')
help_lines = [
"\033[1;36m╔══════════════════════════════════════════════╗\033[0m",
"\033[1;36m║ COMANDI DISPONIBILI ║\033[0m",
"\033[1;36m╠══════════════════════════════════════════════╣\033[0m",
"\033[1;36m║ /help - Mostra questo messaggio ║\033[0m",
"\033[1;36m║ /clear - Pulisci schermo ║\033[0m",
"\033[1;36m║ /exit - Esci ║\033[0m",
"\033[1;36m║ PgUp - Entra in modalità scroll ║\033[0m",
"\033[1;36m║ q - Exit modalità scroll ║\033[0m",
"\033[1;36m╚══════════════════════════════════════════════╝\033[0m",
"\033[1;36m╔════════════════════════════════════════╗\033[0m",
"\033[1;36m║ COMANDI DISPONIBILI ║\033[0m",
"\033[1;36m╠════════════════════════════════════════╣\033[0m",
"\033[1;36m║ /help - Mostra questo messaggio ║\033[0m",
"\033[1;36m║ /clear - Pulisci schermo ║\033[0m",
"\033[1;36m║ /exit - Esci ║\033[0m",
"\033[1;36m║ PgUp - Modalità scroll ║\033[0m",
"\033[1;36m║ q - Exit modalità scroll ║\033[0m",
"\033[1;36m╚════════════════════════════════════════╝\033[0m",
]
for line in help_lines:
self._add_to_history(line)
self.stream.feed(line + '\r\n')
elif result.lower() == '/clear':
self.screen.reset()
self.history.clear()
self.input_buffer = ""
self.cursor_pos = 0
else:
@@ -348,7 +312,6 @@ class TerminalUI:
self._draw()
finally:
# Ripristina terminale
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
except Exception as e:
@@ -360,7 +323,7 @@ class TerminalUI:
sys.stdout.flush()
def close(self):
"""Chiudi il terminale"""
"""Chiudi"""
self.running = False
@@ -375,7 +338,9 @@ def create_terminal_ui(
# Test
if __name__ == "__main__":
def test_callback(text):
return f"Ricevuto: {text}\n\nLunga risposta:\n" + "\n".join([f"Linea {i}" for i in range(1, 21)])
lines = ["Risposta multi-linea:"]
lines += [f"Linea {i}" for i in range(1, 21)]
return "\n".join(lines)
print("Avvio TerminalUI...")
ui = create_terminal_ui()