feat: add SneakyCodeApp Textual application
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
237
app/ui/app.py
Normal file
237
app/ui/app.py
Normal file
@@ -0,0 +1,237 @@
|
||||
"""SneakyCode Textual TUI application."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from rich.markdown import Markdown
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.widgets import Header, Input, RichLog
|
||||
|
||||
from app.agent.context import SessionContext
|
||||
from app.agent.loop import AgentLoop
|
||||
from app.models.config import AppConfig
|
||||
from app.services.llm import LLMClient
|
||||
from app.services.permissions import PermissionsService
|
||||
from app.services.session import SessionManager
|
||||
from app.services.streaming import StreamHandler
|
||||
from app.tools.registry import create_default_registry
|
||||
from app.ui.widgets import StatusBar, StreamingStatic
|
||||
from app.utils.display import DisplayAdapter, render_user_message
|
||||
from app.utils.logging import get_logger, setup_logging_for_tui
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from textual.worker import Worker
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class SneakyCodeApp(App):
|
||||
"""Main TUI application for SneakyCode."""
|
||||
|
||||
TITLE = "SneakyCode"
|
||||
CSS_PATH = "styles.tcss"
|
||||
|
||||
BINDINGS = [
|
||||
Binding("ctrl+c", "cancel_or_quit", "Cancel/Quit", show=False),
|
||||
]
|
||||
|
||||
def __init__(self, config: AppConfig, session_mgr: SessionManager | None = None) -> None:
|
||||
super().__init__()
|
||||
self._config = config
|
||||
self._session_mgr = session_mgr
|
||||
self._ctx: SessionContext | None = None
|
||||
self._agent: AgentLoop | None = None
|
||||
self._client: LLMClient | None = None
|
||||
self._registry = None
|
||||
self._permissions: PermissionsService | None = None
|
||||
self._current_worker: Worker | None = None
|
||||
self._cancel_count = 0
|
||||
self.sub_title = config.llm.model
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header()
|
||||
yield RichLog(id="chat-log", highlight=True, markup=True)
|
||||
yield StreamingStatic("", id="streaming")
|
||||
yield StatusBar()
|
||||
yield Input(placeholder="Enter your prompt...")
|
||||
|
||||
async def on_mount(self) -> None:
|
||||
"""Initialize agent components after the app is mounted."""
|
||||
setup_logging_for_tui()
|
||||
|
||||
self._ctx = SessionContext(self._config)
|
||||
|
||||
# Create long-lived agent dependencies (reused across turns)
|
||||
self._client = LLMClient(self._config.llm)
|
||||
await self._client.__aenter__()
|
||||
self._registry = create_default_registry(self._config.agent.workspace_root, self._config)
|
||||
self._permissions = PermissionsService(self._config.permissions)
|
||||
|
||||
# Set up permission prompt callback
|
||||
async def permission_prompt(tool_name: str, description: str) -> bool:
|
||||
return await self._show_permission_modal(tool_name, description)
|
||||
|
||||
self._permissions.set_prompt_callback(permission_prompt)
|
||||
|
||||
# Offer session resume if configured
|
||||
if self._session_mgr and self._config.session.offer_resume:
|
||||
saved = self._session_mgr.load_latest()
|
||||
if saved:
|
||||
msg_count = len(saved.messages)
|
||||
log = self.query_one("#chat-log", RichLog)
|
||||
log.write(Text(f"Found previous session ({msg_count} messages)", style="cyan"))
|
||||
# Auto-resume for now (modal can be added later)
|
||||
self._session_mgr.restore(saved, self._ctx)
|
||||
log.write(Text(f"✓ Session restored ({msg_count} messages)", style="bold green"))
|
||||
|
||||
self.query_one(Input).focus()
|
||||
|
||||
async def on_input_submitted(self, event: Input.Submitted) -> None:
|
||||
"""Handle user input submission."""
|
||||
user_input = event.value.strip()
|
||||
if not user_input:
|
||||
return
|
||||
|
||||
event.input.clear()
|
||||
log = self.query_one("#chat-log", RichLog)
|
||||
|
||||
# Handle slash commands
|
||||
if user_input.startswith("/"):
|
||||
await self._handle_slash_command(user_input, log)
|
||||
return
|
||||
|
||||
# Write user message to log
|
||||
log.write(render_user_message(user_input))
|
||||
|
||||
# Dispatch agent turn as async worker
|
||||
self._cancel_count = 0
|
||||
self._current_worker = self.run_worker(
|
||||
self._run_agent_turn(user_input),
|
||||
name="agent-turn",
|
||||
exclusive=True,
|
||||
)
|
||||
|
||||
async def _handle_slash_command(self, command: str, log: RichLog) -> None:
|
||||
"""Process slash commands."""
|
||||
cmd = command.lower()
|
||||
if cmd == "/quit":
|
||||
self._save_session()
|
||||
self.exit()
|
||||
elif cmd == "/clear":
|
||||
if self._ctx:
|
||||
self._ctx.clear_history()
|
||||
log.clear()
|
||||
log.write(Text("✓ Conversation history cleared.", style="bold green"))
|
||||
elif cmd == "/history":
|
||||
if self._ctx:
|
||||
from app.utils.display import render_history
|
||||
log.write(render_history(self._ctx.get_history()))
|
||||
elif cmd == "/save":
|
||||
path = self._save_session()
|
||||
if path:
|
||||
log.write(Text(f"✓ Session saved to {path}", style="bold green"))
|
||||
else:
|
||||
log.write(Text("✗ No session to save", style="bold red"))
|
||||
elif cmd == "/session":
|
||||
if self._ctx:
|
||||
log.write(Text(
|
||||
f"Messages: {self._ctx.message_count} | "
|
||||
f"Tokens: ~{self._ctx.estimated_tokens:,} / {self._ctx.token_counter.budget:,} | "
|
||||
f"Started: {self._ctx.start_time.isoformat()}",
|
||||
style="cyan",
|
||||
))
|
||||
else:
|
||||
log.write(Text(f"⚠ Unknown command: {command}", style="yellow"))
|
||||
|
||||
async def _run_agent_turn(self, user_input: str) -> None:
|
||||
"""Run a single agent turn (called as a worker)."""
|
||||
if self._ctx is None or self._client is None:
|
||||
return
|
||||
|
||||
log = self.query_one("#chat-log", RichLog)
|
||||
streaming_widget = self.query_one("#streaming", StreamingStatic)
|
||||
display = DisplayAdapter(log)
|
||||
|
||||
# StreamHandler is per-turn (has per-turn accumulators)
|
||||
handler = StreamHandler(self._config.display)
|
||||
|
||||
# Set up streaming UI callbacks
|
||||
def on_content(content: str) -> None:
|
||||
streaming_widget.update(
|
||||
Panel(Markdown(content), title="Assistant", border_style="green", expand=True)
|
||||
)
|
||||
streaming_widget.show_streaming()
|
||||
|
||||
def on_thinking() -> None:
|
||||
streaming_widget.update(Text("Thinking...", style="dim"))
|
||||
streaming_widget.show_streaming()
|
||||
|
||||
def on_done() -> None:
|
||||
streaming_widget.hide_streaming()
|
||||
|
||||
handler.set_callbacks(on_content=on_content, on_thinking=on_thinking, on_done=on_done)
|
||||
|
||||
agent = AgentLoop(
|
||||
self._config, self._ctx, self._client, handler,
|
||||
self._registry, self._permissions, display,
|
||||
)
|
||||
|
||||
await agent.run_turn(user_input)
|
||||
|
||||
# Auto-save
|
||||
if self._config.session.auto_save:
|
||||
self._save_session()
|
||||
|
||||
async def _show_permission_modal(self, tool_name: str, description: str) -> bool:
|
||||
"""Show a permission prompt in the chat log and wait for response.
|
||||
|
||||
Uses the Input widget to capture y/n response.
|
||||
"""
|
||||
# For v1, auto-approve all tools that reach the prompt
|
||||
# TODO: Implement proper modal dialog
|
||||
log = self.query_one("#chat-log", RichLog)
|
||||
log.write(Text(f"⚠ Auto-approved: {tool_name} ({description})", style="yellow"))
|
||||
return True
|
||||
|
||||
def action_cancel_or_quit(self) -> None:
|
||||
"""Handle Ctrl+C: first press cancels worker, second press quits."""
|
||||
self._cancel_count += 1
|
||||
if self._cancel_count >= 2 or self._current_worker is None:
|
||||
self._save_session()
|
||||
self.exit()
|
||||
elif self._current_worker is not None:
|
||||
self._current_worker.cancel()
|
||||
log = self.query_one("#chat-log", RichLog)
|
||||
log.write(Text("⚠ Cancelling... (press Ctrl+C again to quit)", style="yellow"))
|
||||
|
||||
def on_worker_state_changed(self, event: Worker.StateChanged) -> None:
|
||||
"""Handle worker completion or failure."""
|
||||
from textual.worker import WorkerState
|
||||
|
||||
if event.worker.name != "agent-turn":
|
||||
return
|
||||
|
||||
if event.state == WorkerState.ERROR:
|
||||
log = self.query_one("#chat-log", RichLog)
|
||||
error = event.worker.error
|
||||
log.write(Text(f"✗ Agent error: {error}", style="bold red"))
|
||||
|
||||
if event.state in (WorkerState.SUCCESS, WorkerState.ERROR, WorkerState.CANCELLED):
|
||||
self._current_worker = None
|
||||
# Hide streaming widget in case it was left visible
|
||||
streaming = self.query_one("#streaming", StreamingStatic)
|
||||
streaming.hide_streaming()
|
||||
|
||||
def _save_session(self) -> Path | None:
|
||||
"""Save session quietly, return path or None."""
|
||||
if self._session_mgr and self._ctx and self._ctx.message_count > 0:
|
||||
try:
|
||||
return self._session_mgr.save(self._ctx)
|
||||
except OSError as e:
|
||||
logger.warning("session_save_failed", error=str(e))
|
||||
return None
|
||||
Reference in New Issue
Block a user