diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index 485d73b..2bba038 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -13,7 +13,8 @@ jobs: - uses: actions/setup-python@v5 with: python-version: "3.11" - - name: Install ruff - run: pip install ruff + cache: pip + - name: Install dependencies + run: pip install -e ".[dev]" - name: Run ruff run: ruff check . diff --git a/EvoScientist/cli/__init__.py b/EvoScientist/cli/__init__.py index 9a42d11..bba6054 100644 --- a/EvoScientist/cli/__init__.py +++ b/EvoScientist/cli/__init__.py @@ -13,6 +13,17 @@ from .agent import _deduplicate_run_name # noqa: F401 from ._app import app # noqa: F401 from . import commands # noqa: F401 — registers @app.command decorators +# UI runtime re-exports (merged from former tui/ package) +from .tui_runtime import ( # noqa: F401 + DEFAULT_UI_BACKEND, + SUPPORTED_UI_BACKENDS, + normalize_ui_backend, + resolve_ui_backend, + get_backend, + run_streaming, +) +from ._constants import WELCOME_SLOGANS # noqa: F401 + def main(): """CLI entry point.""" diff --git a/EvoScientist/cli/_constants.py b/EvoScientist/cli/_constants.py new file mode 100644 index 0000000..44f750b --- /dev/null +++ b/EvoScientist/cli/_constants.py @@ -0,0 +1,39 @@ +"""Shared constants and utilities for CLI and TUI modules.""" + +from datetime import datetime, timezone + +from ..sessions import AGENT_NAME + +WELCOME_SLOGANS = [ + "Ready for vibe research? What would you want cooking?", + "Science doesn't sleep. Neither do your sub-agents.", + "From hypothesis to paper — let's cook.", + "Your research kitchen is ready. What's on the menu?", + "Experiments don't run themselves. Oh wait — they do now.", + "Drop a question. We'll bring the citations.", + "Vibe-driven discovery starts here.", + "What breakthrough are we cooking today?", +] + +# ASCII art logo — shared by both Rich CLI and Textual TUI banners. +LOGO_LINES = ( + r" ███████╗ ██╗ ██╗ ██████╗ ███████╗ ██████╗ ██╗ ███████╗ ███╗ ██╗ ████████╗ ██╗ ███████╗ ████████╗", + r" ██╔════╝ ██║ ██║ ██╔═══██╗ ██╔════╝ ██╔════╝ ██║ ██╔════╝ ████╗ ██║ ╚══██╔══╝ ██║ ██╔════╝ ╚══██╔══╝", + r" █████╗ ██║ ██║ ██║ ██║ ███████╗ ██║ ██║ █████╗ ██╔██╗ ██║ ██║ ██║ ███████╗ ██║ ", + r" ██╔══╝ ╚██╗ ██╔╝ ██║ ██║ ╚════██║ ██║ ██║ ██╔══╝ ██║╚██╗██║ ██║ ██║ ╚════██║ ██║ ", + r" ███████╗ ╚████╔╝ ╚██████╔╝ ███████║ ╚██████╗ ██║ ███████╗ ██║ ╚████║ ██║ ██║ ███████║ ██║ ", + r" ╚══════╝ ╚═══╝ ╚═════╝ ╚══════╝ ╚═════╝ ╚═╝ ╚══════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝ ╚══════╝ ╚═╝ ", +) + +# Blue gradient: deep navy -> royal blue -> sky blue -> cyan +LOGO_GRADIENT = ["#1a237e", "#1565c0", "#1e88e5", "#42a5f5", "#64b5f6", "#90caf9"] + + +def build_metadata(workspace_dir: str | None, model: str | None) -> dict: + """Build metadata dict for LangGraph checkpoint persistence.""" + return { + "agent_name": AGENT_NAME, + "updated_at": datetime.now(timezone.utc).isoformat(), + "workspace_dir": workspace_dir or "", + "model": model or "", + } diff --git a/EvoScientist/cli/commands.py b/EvoScientist/cli/commands.py index f9cb972..9a2ddcd 100644 --- a/EvoScientist/cli/commands.py +++ b/EvoScientist/cli/commands.py @@ -408,6 +408,11 @@ def _main_callback( workdir: Optional[str] = typer.Option(None, "--workdir", help="Override workspace directory for this session"), use_cwd: bool = typer.Option(False, "--use-cwd", help="Use current working directory as workspace"), no_thinking: bool = typer.Option(False, "--no-thinking", help="Disable thinking display"), + ui: Optional[str] = typer.Option( + None, + "--ui", + help="UI backend: rich (default) or textual (beta).", + ), ): """EvoScientist Agent - AI-powered research & code execution CLI""" # If a subcommand was invoked, don't run the default behavior @@ -429,6 +434,8 @@ def _main_callback( cli_overrides["default_workdir"] = workdir if no_thinking: cli_overrides["show_thinking"] = False + if ui: + cli_overrides["ui_backend"] = ui config = get_effective_config(cli_overrides) apply_config_to_env(config) @@ -445,6 +452,8 @@ def _main_callback( if mode and mode not in ("run", "daemon"): raise typer.BadParameter("--mode must be 'run' or 'daemon'") + if ui and ui.lower() not in ("rich", "textual"): + raise typer.BadParameter("--ui must be 'rich' or 'textual'") # --name only makes sense in run mode if name and not (mode == "run" or (not mode and not workdir and not use_cwd and config.default_mode == "run")): @@ -523,7 +532,15 @@ def _main_callback( console.print("[dim]Loading agent...[/dim]") agent = _load_agent(workspace_dir=workspace_dir, checkpointer=checkpointer) tid = thread_id or generate_thread_id() - cmd_run(agent, prompt, thread_id=tid, show_thinking=show_thinking, workspace_dir=workspace_dir, model=config.model) + cmd_run( + agent, + prompt, + thread_id=tid, + show_thinking=show_thinking, + workspace_dir=workspace_dir, + model=config.model, + ui_backend=config.ui_backend, + ) import nest_asyncio # type: ignore[import-untyped] nest_asyncio.apply() @@ -540,6 +557,7 @@ def _main_callback( provider=config.provider, run_name=name, thread_id=thread_id, + ui_backend=config.ui_backend, ) diff --git a/EvoScientist/cli/interactive.py b/EvoScientist/cli/interactive.py index 0d59915..2b7caf9 100644 --- a/EvoScientist/cli/interactive.py +++ b/EvoScientist/cli/interactive.py @@ -4,8 +4,8 @@ import asyncio import logging import os import queue +import random import sys -from datetime import datetime, timezone from typing import Any import typer # type: ignore[import-untyped] @@ -29,9 +29,11 @@ from ..sessions import ( get_thread_metadata, get_thread_messages, _format_relative_time, - AGENT_NAME, ) -from ..stream.display import console, _run_streaming +from ..stream.display import console +from .tui_runtime import run_streaming, resolve_ui_backend +from .tui_interactive import run_textual_interactive +from ._constants import WELCOME_SLOGANS, LOGO_LINES, LOGO_GRADIENT, build_metadata from .agent import _shorten_path, _create_session_workspace, _load_agent from .channel import ( ChannelMessage, @@ -53,24 +55,6 @@ _channel_logger = logging.getLogger(__name__) # Banner # ============================================================================= -EVOSCIENTIST_ASCII_LINES = [ - r" \u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2557 \u2588\u2588\u2557 \u2588\u2588\u2557 \u2588\u2588\u2588\u2588\u2588\u2588\u2557 \u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2557 \u2588\u2588\u2588\u2588\u2588\u2588\u2557 \u2588\u2588\u2557 \u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2557 \u2588\u2588\u2588\u2557 \u2588\u2588\u2557 \u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2557 \u2588\u2588\u2557 \u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2557 \u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2588\u2557", -] - -# Blue gradient: deep navy -> royal blue -> sky blue -> cyan -_GRADIENT_COLORS = ["#1a237e", "#1565c0", "#1e88e5", "#42a5f5", "#64b5f6", "#90caf9"] - -# Keep the real ASCII art lines (raw strings) rather than the escaped version above -_REAL_ASCII_LINES = [ - r" ███████╗ ██╗ ██╗ ██████╗ ███████╗ ██████╗ ██╗ ███████╗ ███╗ ██╗ ████████╗ ██╗ ███████╗ ████████╗", - r" ██╔════╝ ██║ ██║ ██╔═══██╗ ██╔════╝ ██╔════╝ ██║ ██╔════╝ ████╗ ██║ ╚══██╔══╝ ██║ ██╔════╝ ╚══██╔══╝", - r" █████╗ ██║ ██║ ██║ ██║ ███████╗ ██║ ██║ █████╗ ██╔██╗ ██║ ██║ ██║ ███████╗ ██║ ", - r" ██╔══╝ ╚██╗ ██╔╝ ██║ ██║ ╚════██║ ██║ ██║ ██╔══╝ ██║╚██╗██║ ██║ ██║ ╚════██║ ██║ ", - r" ███████╗ ╚████╔╝ ╚██████╔╝ ███████║ ╚██████╗ ██║ ███████╗ ██║ ╚████║ ██║ ██║ ███████║ ██║ ", - r" ╚══════╝ ╚═══╝ ╚═════╝ ╚══════╝ ╚═════╝ ╚═╝ ╚══════╝ ╚═╝ ╚═══╝ ╚═╝ ╚═╝ ╚══════╝ ╚═╝ ", -] - - def print_banner( thread_id: str, workspace_dir: str | None = None, @@ -78,29 +62,32 @@ def print_banner( mode: str | None = None, model: str | None = None, provider: str | None = None, + ui_backend: str | None = None, ): - """Print welcome banner with ASCII art logo, thread ID, workspace path, and mode.""" - for line, color in zip(_REAL_ASCII_LINES, _GRADIENT_COLORS): + """Print welcome banner with ASCII art logo, info line, and hint.""" + for line, color in zip(LOGO_LINES, LOGO_GRADIENT): console.print(Text(line, style=f"{color} bold")) info = Text() - if model or provider or mode: - info.append(" ", style="dim") - parts = [] - if model: - parts.append(("Model: ", model)) - if provider: - parts.append(("Provider: ", provider)) - if mode: - parts.append(("Mode: ", mode)) - for i, (label, value) in enumerate(parts): - if i > 0: - info.append(" ", style="dim") - info.append(label, style="dim") - info.append(value, style="magenta") + info.append(" ", style="dim") + parts: list[tuple[str, str]] = [] + if model: + parts.append(("Model: ", model)) + if provider: + parts.append(("Provider: ", provider)) + if mode: + parts.append(("Mode: ", mode)) + if ui_backend: + parts.append(("UI: ", ui_backend)) + for i, (label, value) in enumerate(parts): + if i > 0: + info.append(" ", style="dim") + info.append(label, style="dim") + info.append(value, style="magenta") info.append("\n Type ", style="#ffe082") info.append("/", style="#ffe082 bold") info.append(" for commands", style="#ffe082") console.print(info) + console.print(Text(f" {random.choice(WELCOME_SLOGANS)}", style="dim italic")) console.print() @@ -166,16 +153,6 @@ class SlashCommandCompleter(Completer): # ============================================================================= -def _build_metadata(workspace_dir: str | None, model: str | None) -> dict: - """Build metadata dict for LangGraph checkpoint persistence.""" - return { - "agent_name": AGENT_NAME, - "updated_at": datetime.now(timezone.utc).isoformat(), - "workspace_dir": workspace_dir or "", - "model": model or "", - } - - def cmd_interactive( show_thinking: bool = True, channel_send_thinking: bool = True, @@ -186,6 +163,7 @@ def cmd_interactive( provider: str | None = None, run_name: str | None = None, thread_id: str | None = None, + ui_backend: str = "rich", ) -> None: """Interactive conversation mode with streaming output. @@ -202,10 +180,28 @@ def cmd_interactive( provider: LLM provider name to display in banner run_name: Optional run name for /new session deduplication thread_id: Optional thread ID to resume a previous session + ui_backend: UI backend ('rich' or 'textual') """ import nest_asyncio nest_asyncio.apply() + resolved_ui_backend = resolve_ui_backend(ui_backend, warn_fallback=True) + if resolved_ui_backend == "textual": + run_textual_interactive( + show_thinking=show_thinking, + channel_send_thinking=channel_send_thinking, + workspace_dir=workspace_dir, + workspace_fixed=workspace_fixed, + mode=mode, + model=model, + provider=provider, + run_name=run_name, + thread_id=thread_id, + load_agent=_load_agent, + create_session_workspace=_create_session_workspace, + ) + return + from .. import paths memory_dir = str(paths.MEMORY_DIR) @@ -231,6 +227,7 @@ def cmd_interactive( "workspace_dir": workspace_dir, "running": True, "resumed": False, + "ui_backend": resolved_ui_backend, } async def _resolve_thread_id(tid: str) -> str | None: @@ -433,10 +430,26 @@ def cmd_interactive( # Print banner if state["resumed"]: - print_banner(state["thread_id"], state["workspace_dir"], memory_dir, mode, model, provider) + print_banner( + state["thread_id"], + state["workspace_dir"], + memory_dir, + mode, + model, + provider, + state["ui_backend"], + ) console.print(f"[green]Resumed session [yellow]{state['thread_id']}[/yellow][/green]\n") else: - print_banner(state["thread_id"], state["workspace_dir"], memory_dir, mode, model, provider) + print_banner( + state["thread_id"], + state["workspace_dir"], + memory_dir, + mode, + model, + provider, + state["ui_backend"], + ) # ---- Channel queue processing (bus → main thread) ---- @@ -504,11 +517,16 @@ def cmd_interactive( "Media", timeout=30, ) - meta = _build_metadata(state["workspace_dir"], model) + meta = build_metadata(state["workspace_dir"], model) try: - response = _run_streaming( - state["agent"], msg.content, state["thread_id"], - show_thinking, interactive=True, metadata=meta, + response = run_streaming( + ui_backend=state["ui_backend"], + agent=state["agent"], + message=msg.content, + thread_id=state["thread_id"], + show_thinking=show_thinking, + interactive=True, + metadata=meta, on_thinking=_send_thinking_to_channel, on_todo=_send_todo_to_channel, on_file_write=_send_media_to_channel, @@ -611,6 +629,7 @@ def cmd_interactive( console.print(f"[dim]Thread:[/dim] [yellow]{state['thread_id']}[/yellow]") if state["workspace_dir"]: console.print(f"[dim]Workspace:[/dim] [cyan]{_shorten_path(state['workspace_dir'])}[/cyan]") + console.print(f"[dim]UI:[/dim] [cyan]{state['ui_backend']}[/cyan]") if memory_dir: console.print(f"[dim]Memory dir:[/dim] [cyan]{_shorten_path(memory_dir)}[/cyan]") console.print() @@ -650,10 +669,15 @@ def cmd_interactive( # Stream agent response with metadata for persistence console.print() - meta = _build_metadata(state["workspace_dir"], model) - _run_streaming( - state["agent"], user_input, state["thread_id"], - show_thinking, interactive=True, metadata=meta, + meta = build_metadata(state["workspace_dir"], model) + run_streaming( + ui_backend=state["ui_backend"], + agent=state["agent"], + message=user_input, + thread_id=state["thread_id"], + show_thinking=show_thinking, + interactive=True, + metadata=meta, ) console.print() _print_separator() @@ -697,6 +721,7 @@ def cmd_run( show_thinking: bool = True, workspace_dir: str | None = None, model: str | None = None, + ui_backend: str = "rich", ) -> None: """Single-shot execution with streaming display. @@ -707,6 +732,7 @@ def cmd_run( show_thinking: Whether to display thinking panels workspace_dir: Per-session workspace directory path model: Model name for checkpoint metadata + ui_backend: UI backend ('rich' or 'textual') """ thread_id = thread_id or generate_thread_id() @@ -720,9 +746,17 @@ def cmd_run( console.print(f"[dim]Workspace: {_shorten_path(workspace_dir)}[/dim]") console.print() - meta = _build_metadata(workspace_dir, model) + meta = build_metadata(workspace_dir, model) try: - _run_streaming(agent, prompt, thread_id, show_thinking, interactive=False, metadata=meta) + run_streaming( + ui_backend=resolve_ui_backend(ui_backend, warn_fallback=True), + agent=agent, + message=prompt, + thread_id=thread_id, + show_thinking=show_thinking, + interactive=False, + metadata=meta, + ) except Exception as e: error_msg = str(e) if "authentication" in error_msg.lower() or "api_key" in error_msg.lower(): diff --git a/EvoScientist/cli/tui_backends.py b/EvoScientist/cli/tui_backends.py new file mode 100644 index 0000000..53db358 --- /dev/null +++ b/EvoScientist/cli/tui_backends.py @@ -0,0 +1,61 @@ +"""TUI backend abstractions for streaming output.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Callable, Protocol + +from ..stream.display import _run_streaming + + +class StreamingTUIBackend(Protocol): + """Protocol for TUI backends that can render agent streaming output.""" + + name: str + + def run_streaming( + self, + *, + agent: Any, + message: str, + thread_id: str, + show_thinking: bool, + interactive: bool, + on_thinking: Callable[[str], None] | None = None, + on_todo: Callable[[list[dict]], None] | None = None, + on_file_write: Callable[[str], None] | None = None, + metadata: dict | None = None, + ) -> str: + """Run streaming and return final response text.""" + + +@dataclass(slots=True) +class RichStreamingBackend: + """Default Rich backend wrapper around the existing streaming renderer.""" + + name: str = "rich" + + def run_streaming( + self, + *, + agent: Any, + message: str, + thread_id: str, + show_thinking: bool, + interactive: bool, + on_thinking: Callable[[str], None] | None = None, + on_todo: Callable[[list[dict]], None] | None = None, + on_file_write: Callable[[str], None] | None = None, + metadata: dict | None = None, + ) -> str: + return _run_streaming( + agent=agent, + message=message, + thread_id=thread_id, + show_thinking=show_thinking, + interactive=interactive, + on_thinking=on_thinking, + on_todo=on_todo, + on_file_write=on_file_write, + metadata=metadata, + ) diff --git a/EvoScientist/cli/tui_interactive.py b/EvoScientist/cli/tui_interactive.py new file mode 100644 index 0000000..9cb8c1e --- /dev/null +++ b/EvoScientist/cli/tui_interactive.py @@ -0,0 +1,1795 @@ +"""Full-screen Textual interactive TUI for EvoScientist. + +Widget-based rendering: each message/tool/sub-agent is an independent widget +mounted into a VerticalScroll container. No timer-based Group rebuilds. +""" + +from __future__ import annotations + +import asyncio +import logging +import queue +import shlex +from pathlib import Path +from typing import Any, Callable + +from rich.console import Group +from rich.table import Table +from rich.text import Text + +import EvoScientist.cli.channel as _ch_mod +from .channel import ( + ChannelMessage, + _channels_is_running, + _channels_running_list, + _channels_stop, + _auto_start_channel, + _message_queue, + _set_channel_response, +) +from ..sessions import ( + _format_relative_time, + delete_thread, + find_similar_threads, + generate_thread_id, + get_checkpointer, + get_thread_messages, + get_thread_metadata, + list_threads, + thread_exists, +) +from ..stream.events import stream_agent_events +from ..stream.state import StreamState, _INTERNAL_TOOLS + +from ._constants import LOGO_LINES, LOGO_GRADIENT, build_metadata + +_channel_logger = logging.getLogger(__name__) + +_TUI_SLASH_COMMANDS = [ + ("/current", "Show current session info"), + ("/threads", "List recent sessions"), + ("/resume", "Resume a previous session"), + ("/delete", "Delete a saved session"), + ("/new", "Start a new session"), + ("/clear", "Clear chat history"), + ("/skills", "List installed skills"), + ("/install-skill", "Add a skill from path or GitHub"), + ("/uninstall-skill", "Remove an installed skill"), + ("/mcp", "Manage MCP servers"), + ("/channel", "Configure messaging channels"), + ("/help", "Show available commands"), + ("/exit", "Quit EvoScientist"), +] + + +def _home_relative_path(path_str: str | None) -> str: + """Render paths in a compact, user-friendly form.""" + if not path_str: + return "" + path = Path(path_str).expanduser().resolve() + home = Path.home().resolve() + try: + if path.is_relative_to(home): + return f"~/{path.relative_to(home)}" + except ValueError: + pass + return str(path) + + +def _build_welcome_banner( + *, + thread_id: str, + workspace_dir: str | None, + mode: str | None, + model: str | None, + provider: str | None, + ui_backend: str | None = None, + channels: list[tuple[str, bool, str]] | None = None, +) -> Any: + """Build CLI-matching welcome banner with logo, info line, and channels. + + Args: + channels: List of (name, ok, detail) tuples for the channels panel. + """ + banner = Text() + for line, color in zip(LOGO_LINES, LOGO_GRADIENT): + banner.append(f"{line}\n", style=f"bold {color}") + + # Info line — matches CLI print_banner format + info = Text() + parts: list[tuple[str, str]] = [] + if model: + parts.append(("Model: ", model)) + if provider: + parts.append(("Provider: ", provider)) + if mode: + parts.append(("Mode: ", mode)) + if ui_backend: + parts.append(("UI: ", ui_backend)) + if parts: + info.append(" ", style="dim") + for i, (label, value) in enumerate(parts): + if i > 0: + info.append(" ", style="dim") + info.append(label, style="dim") + info.append(value, style="magenta") + info.append("\n Type ", style="#ffe082") + info.append("/", style="#ffe082 bold") + info.append(" for commands", style="#ffe082") + banner.append_text(info) + + # Channels panel + if channels: + from rich.panel import Panel + + lines: list[Text] = [] + all_ok = True + for name, ok, detail in channels: + line = Text() + if ok: + line.append("\u25cf ", style="green") + line.append(name, style="bold") + else: + line.append("\u25cb ", style="dim") + line.append(name, style="bold dim") + all_ok = False + if detail: + line.append(f" {detail}", style="dim") + lines.append(line) + body = Text("\n").join(lines) + border = "green" if all_ok else "dim" + panel = Panel(body, title="[bold]Channels[/bold]", border_style=border, expand=False) + return Group(banner, Text(""), panel) + + return banner + + +def _is_final_response(state: StreamState) -> bool: + """Check if all tools are done and no sub-agents are active.""" + n_visible = 0 + n_done = 0 + for i, tc in enumerate(state.tool_calls): + if tc.get("name") in _INTERNAL_TOOLS: + continue + n_visible += 1 + if i < len(state.tool_results): + n_done += 1 + has_pending = n_visible > n_done + any_active_sa = any(sa.is_active for sa in state.subagents) + return not has_pending and not any_active_sa and not state.is_processing + + +def run_textual_interactive( + *, + show_thinking: bool, + channel_send_thinking: bool = True, + workspace_dir: str | None, + workspace_fixed: bool, + mode: str | None, + model: str | None, + provider: str | None, + run_name: str | None, + thread_id: str | None, + load_agent: Callable[..., Any], + create_session_workspace: Callable[[str | None], str], +) -> None: + """Run full-screen Textual interactive chat loop.""" + try: + from textual.app import App, ComposeResult + from textual.binding import Binding + from textual.containers import Container, Horizontal, VerticalScroll + from textual.widgets import Input, Static + + from .widgets import ( + LoadingWidget, + ThinkingWidget, + AssistantMessage, + ToolCallWidget, + SubAgentWidget, + TodoWidget, + UserMessage, + SystemMessage, + ) + except Exception as e: # pragma: no cover - runtime fallback path + raise RuntimeError( + "Textual TUI backend requires 'textual'. Run: pip install textual" + ) from e + + class EvoTextualInteractiveApp(App[None]): # type: ignore[type-arg] + """Deep-Agents-style full-screen TUI with independent widget rendering.""" + + CSS = """ + Screen { + layout: vertical; + background: #06080d; + color: #d1d5db; + } + #chat { + height: 1fr; + padding: 1 2; + background: #06080d; + } + #welcome { + height: auto; + margin-bottom: 1; + } + #input-shell { + height: auto; + padding: 0 2 1 2; + background: #06080d; + } + #input-row { + height: 3; + border: solid #0284c7; + background: #11151b; + padding: 0 1; + } + #input-cursor { + width: 2; + content-align: center middle; + color: #0284c7; + text-style: bold; + } + #prompt { + width: 1fr; + border: none; + background: transparent; + color: #e5e7eb; + } + #prompt:focus { + border: none; + } + #completions { + display: none; + height: auto; + max-height: 15; + background: #11151b; + padding: 0 1; + border-bottom: solid #0284c7; + } + #status { + height: 1; + background: #171a20; + color: #f59e0b; + padding: 0 1; + } + """ + BINDINGS = [ + Binding("ctrl+c", "request_quit", "Quit", show=False), + ] + + def __init__( + self, + *, + agent: Any, + thread_id_value: str, + workspace: str | None, + checkpointer: Any, + channel_send_thinking_value: bool = True, + resumed: bool = False, + ) -> None: + super().__init__() + self._agent = agent + self._thread_id = thread_id_value + self._workspace_dir = workspace + self._checkpointer = checkpointer + self._channel_send_thinking = channel_send_thinking_value + self._resumed = resumed + self._channel_timer: Any = None + self._started_channel_types: list[str] = [] + self._busy = False + self._run_task: Any = None # asyncio.Task for current _run_turn + self._comp_items: list[tuple[str, str]] = [] + self._comp_index: int = -1 + + # ── Layout ───────────────────────────────────────────── + + def compose(self) -> ComposeResult: + with VerticalScroll(id="chat"): + yield Static("", id="welcome") + # Widgets are mounted directly here by _stream_with_widgets, + # _append_system, _mount_renderable, etc. + + with Container(id="input-shell"): + yield Static("", id="completions") + with Horizontal(id="input-row"): + yield Static(">", id="input-cursor") + yield Input( + placeholder="Type message (/ for commands)", + id="prompt", + ) + + yield Static("", id="status") + + def on_mount(self) -> None: + self._render_welcome() + self._render_status() + self.query_one("#prompt", Input).focus() + # Show resumed session info (matches CLI behavior) + if self._resumed: + self._append_system( + f"Resumed session: {self._thread_id}", style="green", + ) + self.call_later( + lambda: asyncio.ensure_future(self._render_history(self._thread_id)) + ) + # Auto-start channels + self._start_channels() + + # ── Channel integration ──────────────────────────────── + + def _start_channels(self) -> None: + """Auto-start channels if enabled in config.""" + try: + from ..config import load_config + + cfg = load_config() + if cfg and cfg.channel_enabled and not _channels_is_running(): + _auto_start_channel( + self._agent, + self._thread_id, + cfg, + send_thinking=self._channel_send_thinking, + ) + types = [ + t.strip() + for t in cfg.channel_enabled.split(",") + if t.strip() + ] + self._started_channel_types = types + self._render_welcome() + except Exception as e: + _channel_logger.debug(f"Channel auto-start failed: {e}") + self._channel_timer = self.set_interval(0.1, self._poll_channel_queue) + + def _poll_channel_queue(self) -> None: + """Poll the channel message queue (called every 100ms).""" + try: + msg = _message_queue.get_nowait() + except queue.Empty: + return + if self._busy: + _message_queue.put(msg) + return + self.call_later(lambda m=msg: asyncio.ensure_future(self._process_channel_message(m))) + + # ── Widget helpers ───────────────────────────────────── + + def _append_system(self, text: str, *, style: str = "dim") -> None: + """Mount a SystemMessage widget into #chat.""" + container = self.query_one("#chat", VerticalScroll) + container.mount(SystemMessage(text, msg_style=style)) + container.scroll_end(animate=False) + + def _mount_renderable(self, renderable: Any) -> None: + """Mount a Rich renderable (e.g. Table) as a Static widget.""" + container = self.query_one("#chat", VerticalScroll) + container.mount(Static(renderable)) + container.scroll_end(animate=False) + + # ── Streaming core ───────────────────────────────────── + + async def _stream_with_widgets( + self, + user_text: str, + *, + on_thinking_cb: Callable[[str], None] | None = None, + on_todo_cb: Callable[[list[dict]], None] | None = None, + on_media_cb: Callable[[str], None] | None = None, + ) -> str: + """Stream agent events and mount widgets. Returns response text. + + Shared by ``_run_turn`` (interactive) and + ``_process_channel_message`` (channel). + """ + container = self.query_one("#chat", VerticalScroll) + + # 1. Mount user message + loading spinner + await container.mount(UserMessage(user_text)) + loading = LoadingWidget() + await container.mount(loading) + container.scroll_end(animate=False) + + # 2. Event-driven widget rendering + state = StreamState() + loading_removed = False + thinking_w: ThinkingWidget | None = None + assistant_w: AssistantMessage | None = None + todo_w: TodoWidget | None = None + tool_widgets: dict[str, ToolCallWidget] = {} + subagent_widgets: dict[str, SubAgentWidget] = {} + + # Transient indicator widgets (auto-removed on state transitions) + narration_w: Static | None = None # dim italic intermediate text + processing_w: Static | None = None # "Analyzing results..." + + # Tool collapsing (matches CLI MAX_VISIBLE_TOOLS) + _MAX_VISIBLE_TOOLS = 4 + completed_tool_order: list[str] = [] # tool_ids in completion order + collapse_summary_w: Static | None = None + has_used_tools = False + + _thinking_sent = False + _todo_sent = False + _media_sent: set[str] = set() + _MIN_THINKING_LEN = 200 + + metadata = build_metadata(self._workspace_dir, model) + response = "" + + async def _remove_w(w: Static | None) -> None: + """Safely remove a transient indicator widget.""" + if w is not None: + try: + await w.remove() + except Exception: + pass + + async def _collapse_completed_tools() -> None: + """Hide older completed tool widgets; show summary line.""" + nonlocal collapse_summary_w + completed = [ + (tid, tool_widgets[tid]) + for tid in completed_tool_order + if tid in tool_widgets + ] + n = len(completed) + if n <= _MAX_VISIBLE_TOOLS: + if collapse_summary_w is not None: + collapse_summary_w.display = False + return + + to_hide = n - _MAX_VISIBLE_TOOLS + ok_count = 0 + fail_count = 0 + for i, (_, tw) in enumerate(completed): + if i < to_hide: + tw.display = False + if tw._status == "success": + ok_count += 1 + else: + fail_count += 1 + else: + tw.display = True + + summary = Text() + summary.append(f"\u2713 {ok_count} completed", style="dim green") + if fail_count > 0: + summary.append(f" | {fail_count} failed", style="dim red") + + if collapse_summary_w is None: + collapse_summary_w = Static(summary) + # Position before first visible tool widget + first_visible = None + for _, tw in completed[to_hide:]: + if tw.display: + first_visible = tw + break + if first_visible: + await container.mount(collapse_summary_w, before=first_visible) + else: + await container.mount(collapse_summary_w) + else: + collapse_summary_w.update(summary) + collapse_summary_w.display = True + + def _find_or_rename_sa_widget( + resolved_name: str, description: str = "", + ) -> SubAgentWidget | None: + """Look up a sub-agent widget, renaming 'sub-agent' entry if needed.""" + if resolved_name in subagent_widgets: + w = subagent_widgets[resolved_name] + if description and not w._description: + w.update_name(w._sa_name, description) + return w + # Rename "sub-agent" → real name (mirrors state._get_or_create_subagent) + if resolved_name != "sub-agent" and "sub-agent" in subagent_widgets: + w = subagent_widgets.pop("sub-agent") + w.update_name(resolved_name, description) + subagent_widgets[resolved_name] = w + return w + return None + + try: + async for event in stream_agent_events( + self._agent, + user_text, + self._thread_id, + metadata=metadata, + ): + event_type = state.handle_event(event) + + # -- Channel callbacks (thinking, todo, media) -- + if ( + on_thinking_cb + and not _thinking_sent + and state.thinking_text + and event_type != "thinking" + and len(state.thinking_text) >= _MIN_THINKING_LEN + ): + on_thinking_cb(state.thinking_text.rstrip()) + _thinking_sent = True + + if ( + on_todo_cb + and not _todo_sent + and event_type == "tool_call" + and event.get("name") == "write_todos" + and state.todo_items + ): + if ( + on_thinking_cb + and not _thinking_sent + and state.thinking_text + and len(state.thinking_text) >= _MIN_THINKING_LEN + ): + on_thinking_cb(state.thinking_text.rstrip()) + _thinking_sent = True + on_todo_cb(state.todo_items) + _todo_sent = True + + if ( + on_media_cb + and event_type == "tool_result" + and event.get("success") + ): + tool_name = event.get("name", "") + if tool_name in ("write_file", "read_file"): + _forward_media_to_channel( + state, tool_name, _media_sent, on_media_cb, + ) + + # -- Remove loading spinner on first content event -- + if not loading_removed and event_type in ( + "thinking", "text", "tool_call", + ): + await loading.cleanup() + loading_removed = True + + # -- Widget dispatch -- + if event_type == "thinking": + if thinking_w is None: + thinking_w = ThinkingWidget(show_thinking=show_thinking) + await container.mount(thinking_w) + thinking_w.append_text(event.get("content", "")) + + elif event_type == "text": + if thinking_w is not None and thinking_w._is_active: + thinking_w.finalize() + # Clear processing indicator + await _remove_w(processing_w) + processing_w = None + + if has_used_tools and not _is_final_response(state): + # Tools still running — show intermediate narration + await _remove_w(narration_w) + narration_w = None + last_line = state.latest_text.strip().split("\n")[-1].strip() + if last_line: + if len(last_line) > 60: + last_line = last_line[:57] + "\u2026" + narration_w = Static( + Text(f" {last_line}", style="dim italic"), + ) + await container.mount(narration_w) + else: + # Stream final response incrementally (both + # text-only replies and post-tool responses). + await _remove_w(narration_w) + narration_w = None + if assistant_w is None: + assistant_w = AssistantMessage(state.response_text) + await container.mount(assistant_w) + else: + await assistant_w.append_content( + event.get("content", ""), + ) + + elif event_type == "tool_call": + tool_name = event.get("name", "unknown") + tool_id = event.get("id", "") + tool_args = event.get("args", {}) + # Finalize thinking if still active + if thinking_w is not None and thinking_w._is_active: + thinking_w.finalize() + # Clear transient indicators + await _remove_w(narration_w) + narration_w = None + await _remove_w(processing_w) + processing_w = None + # Remove early AssistantMessage (text arrived before tools) + if assistant_w is not None: + try: + await assistant_w.remove() + except Exception: + pass + assistant_w = None + # Skip internal tools and task (handled by SubAgentWidget) + if tool_name not in _INTERNAL_TOOLS and tool_name != "task": + has_used_tools = True + if tool_id and tool_id in tool_widgets: + # Re-emitted with updated args — update in place + existing = tool_widgets[tool_id] + existing._tool_name = tool_name + existing._tool_args = tool_args + try: + existing._render_header() + except Exception: + pass + else: + w = ToolCallWidget(tool_name, tool_args, tool_id) + await container.mount(w) + if tool_id: + tool_widgets[tool_id] = w + # Update todo widget on write_todos + if tool_name == "write_todos" and state.todo_items: + if todo_w is None: + todo_w = TodoWidget(state.todo_items) + await container.mount(todo_w) + else: + todo_w.update_items(state.todo_items) + + elif event_type == "tool_result": + result_name = event.get("name", "unknown") + result_content = event.get("content", "") + result_success = event.get("success", True) + # Match via state's deduplicated tool_calls (uses tool_id) + matched = False + matched_tid = "" + result_idx = len(state.tool_results) - 1 + if 0 <= result_idx < len(state.tool_calls): + tc = state.tool_calls[result_idx] + tid = tc.get("id", "") + if tid and tid in tool_widgets: + tw = tool_widgets[tid] + if tw._status == "running": + if result_success: + tw.set_success(result_content) + else: + tw.set_error(result_content) + matched = True + matched_tid = tid + # Fallback: match first running widget with same name + if not matched: + for fid, tw in tool_widgets.items(): + if tw.tool_name == result_name and tw._status == "running": + if result_success: + tw.set_success(result_content) + else: + tw.set_error(result_content) + matched = True + matched_tid = fid + break + # Track completion order for collapsing + if matched_tid and matched_tid not in completed_tool_order: + completed_tool_order.append(matched_tid) + await _collapse_completed_tools() + # Update todo from results + if result_name in ("write_todos", "read_todos") and state.todo_items: + if todo_w is None: + todo_w = TodoWidget(state.todo_items) + await container.mount(todo_w) + else: + todo_w.update_items(state.todo_items) + # Show "Analyzing results..." if all tools done, no text yet + if ( + _is_final_response(state) + and not state.response_text + and processing_w is None + ): + processing_w = Static( + Text("\u25cf Analyzing results...", style="cyan"), + ) + await container.mount(processing_w) + + elif event_type == "subagent_start": + sa_name = event.get("name", "sub-agent") + sa_desc = event.get("description", "") + existing = _find_or_rename_sa_widget(sa_name, sa_desc) + if existing is None: + sa_w = SubAgentWidget(sa_name, sa_desc) + await container.mount(sa_w) + subagent_widgets[sa_name] = sa_w + + elif event_type == "subagent_tool_call": + sa_name = event.get("subagent", "sub-agent") + sa_name = state._resolve_subagent_name(sa_name) + sa_w = _find_or_rename_sa_widget(sa_name) + if sa_w is None: + sa_w = SubAgentWidget(sa_name) + await container.mount(sa_w) + subagent_widgets[sa_name] = sa_w + await sa_w.add_tool_call( + event.get("name", "unknown"), + event.get("args", {}), + event.get("id", ""), + ) + + elif event_type == "subagent_tool_result": + sa_name = event.get("subagent", "sub-agent") + sa_name = state._resolve_subagent_name(sa_name) + sa_w = _find_or_rename_sa_widget(sa_name) + if sa_w is not None: + sa_w.complete_tool( + event.get("name", "unknown"), + event.get("content", ""), + event.get("success", True), + ) + + elif event_type == "subagent_end": + sa_name = event.get("name", "sub-agent") + sa_name = state._resolve_subagent_name(sa_name) + sa_w = _find_or_rename_sa_widget(sa_name) + if sa_w is not None: + sa_w.finalize() + + elif event_type == "done": + # Clean up transient indicators + await _remove_w(narration_w) + narration_w = None + await _remove_w(processing_w) + processing_w = None + # Mount final response + if assistant_w is None and state.response_text: + # Strip trailing standalone "..." + clean = state.response_text.strip() + while clean.endswith("\n...") or clean.rstrip() == "...": + clean = clean.rstrip().removesuffix("...").rstrip() + assistant_w = AssistantMessage(clean or state.response_text) + await container.mount(assistant_w) + # Markdown rendering is async and needs multiple + # layout cycles to compute final height. Schedule + # extra deferred scrolls so the response is visible. + self.set_timer( + 0.15, + lambda: container.scroll_end(animate=False), + ) + self.set_timer( + 0.4, + lambda: container.scroll_end(animate=False), + ) + + elif event_type == "error": + error_msg = event.get("message", "Unknown error") + self._append_system(f"Error: {error_msg}", style="red") + + # Scroll after Textual processes the layout update + self.call_after_refresh( + container.scroll_end, animate=False, + ) + + response = (state.response_text or "").strip() + + except asyncio.CancelledError: + # Ctrl+C cancellation + pass + except Exception as exc: + error_msg = str(exc) + if "authentication" in error_msg.lower() or "api_key" in error_msg.lower(): + self._append_system( + "Error: API key not configured.", style="red", + ) + self._append_system( + "Run EvoSci onboard to set up your API key.", style="dim", + ) + else: + self._append_system(f"Error: {exc}", style="red") + response = f"Error: {exc}" + finally: + # Clean up loading widget if it wasn't removed yet + if not loading_removed: + try: + await loading.cleanup() + except Exception: + pass + # Clean up transient indicators + for w in (narration_w, processing_w): + await _remove_w(w) + # Stop timers on any remaining running tool widgets + for tw in tool_widgets.values(): + if tw._status == "running": + tw._stop_timer() + tw._status = "success" + try: + tw._render_header() + tw._render_status() + except Exception: + pass + # Finalize any still-active sub-agents + for sa_w in subagent_widgets.values(): + if sa_w._is_active: + try: + sa_w.finalize() + except Exception: + pass + # Finalize thinking widget + if thinking_w is not None and thinking_w._is_active: + try: + thinking_w.finalize() + except Exception: + pass + # Finalize assistant message stream + if assistant_w is not None: + await assistant_w.stop_stream() + # Flush remaining thinking callback + if ( + on_thinking_cb + and not _thinking_sent + and state.thinking_text + and len(state.thinking_text) >= _MIN_THINKING_LEN + ): + on_thinking_cb(state.thinking_text.rstrip()) + # Final scroll to ensure last content is visible + self.call_after_refresh( + container.scroll_end, animate=False, + ) + + return response + + async def _run_turn(self, user_text: str) -> None: + """Handle a user turn: stream agent response with widgets.""" + self._busy = True + self._render_status() + + prompt = self.query_one("#prompt", Input) + prompt.disabled = True + + try: + await self._stream_with_widgets(user_text) + finally: + self._busy = False + self._run_task = None + self._render_status() + prompt.disabled = False + prompt.focus() + + async def _process_channel_message(self, msg: ChannelMessage) -> None: + """Process a channel message: stream agent response and reply.""" + self._busy = True + self._render_status() + + prompt_widget = self.query_one("#prompt", Input) + prompt_widget.disabled = True + + self._append_system( + f"[{msg.channel_type}: Received from {msg.sender}]", + style="dim", + ) + + # Build channel callbacks + def _send_to_channel(coro, label: str, timeout: int = 15) -> None: + loop = _ch_mod._bus_loop + if not loop: + return + try: + asyncio.run_coroutine_threadsafe(coro, loop).result(timeout=timeout) + except Exception as e: + _channel_logger.debug(f"{label} send failed: {e}") + + def _send_thinking(thinking: str) -> None: + ch = msg.channel_ref + if ch and ch.send_thinking: + _send_to_channel( + ch.send_thinking_message( + sender=msg.chat_id, thinking=thinking, metadata=msg.metadata, + ), + "Thinking", + ) + + def _send_todo(items: list[dict]) -> None: + from ..channels.consumer import _format_todo_list + + if msg.channel_ref: + _send_to_channel( + msg.channel_ref.send_todo_message( + sender=msg.chat_id, + content=_format_todo_list(items), + metadata=msg.metadata, + ), + "Todo", + ) + + def _send_media(file_path: str) -> None: + if msg.channel_ref: + _send_to_channel( + msg.channel_ref.send_media( + recipient=msg.chat_id, + file_path=file_path, + metadata=msg.metadata, + ), + "Media", + timeout=30, + ) + + response = "" + try: + response = await self._stream_with_widgets( + msg.content, + on_thinking_cb=_send_thinking if self._channel_send_thinking else None, + on_todo_cb=_send_todo, + on_media_cb=_send_media, + ) + except Exception as exc: + response = f"Error: {exc}" + self._append_system(f"Error: {exc}", style="red") + finally: + self._busy = False + self._render_status() + prompt_widget.disabled = False + prompt_widget.focus() + + _set_channel_response(msg.msg_id, response) + self._append_system( + f"[{msg.channel_type}: Replied to {msg.sender}]", + style="dim", + ) + + # ── Input handling ───────────────────────────────────── + + async def on_input_submitted(self, event: Input.Submitted) -> None: + text = event.value.strip() + prompt = self.query_one("#prompt", Input) + prompt.value = "" + if not text: + return + + if self._busy: + self._append_system( + "A response is still streaming. Please wait.", + style="yellow", + ) + return + + if text.startswith("/"): + self._hide_completions() + await self._handle_command(text) + return + + self._run_task = asyncio.ensure_future(self._run_turn(text)) + + def on_input_changed(self, event: Input.Changed) -> None: + text = event.value + comp_widget = self.query_one("#completions", Static) + if text.startswith("/"): + prefix = text.lower() + matches = [ + (cmd, desc) + for cmd, desc in _TUI_SLASH_COMMANDS + if cmd.startswith(prefix) + ] + if len(matches) == 1 and matches[0][0] == prefix: + self._hide_completions() + return + if matches: + self._comp_items = matches + self._comp_index = -1 + self._render_completions() + comp_widget.display = True + return + self._hide_completions() + + def on_key(self, event: Any) -> None: + comp_widget = self.query_one("#completions", Static) + if not (comp_widget.display and self._comp_items): + return + + if event.key in ("tab", "down"): + event.prevent_default() + event.stop() + self._comp_index = (self._comp_index + 1) % len(self._comp_items) + self._apply_selected_completion() + elif event.key == "up": + event.prevent_default() + event.stop() + self._comp_index = (self._comp_index - 1) % len(self._comp_items) + self._apply_selected_completion() + elif event.key == "enter" and self._comp_index >= 0: + event.prevent_default() + event.stop() + self._hide_completions() + + def _apply_selected_completion(self) -> None: + selected_cmd = self._comp_items[self._comp_index][0] + prompt = self.query_one("#prompt", Input) + prompt.value = selected_cmd + " " + prompt.cursor_position = len(prompt.value) + self._render_completions() + + def _hide_completions(self) -> None: + self._comp_items = [] + self._comp_index = -1 + comp_widget = self.query_one("#completions", Static) + comp_widget.display = False + + def _render_completions(self) -> None: + comp_text = Text() + for i, (cmd, desc) in enumerate(self._comp_items): + if i == self._comp_index: + comp_text.append("\u25b8 ", style="bold") + comp_text.append(f"{cmd:<22}", style="bold") + comp_text.append(desc, style="bold") + else: + comp_text.append(" ", style="#888888") + comp_text.append(f"{cmd:<22}", style="#888888") + comp_text.append(desc, style="#888888") + if i < len(self._comp_items) - 1: + comp_text.append("\n") + self.query_one("#completions", Static).update(comp_text) + + # ── Slash commands ───────────────────────────────────── + + async def _handle_command(self, command: str) -> None: + cmd, _, arg = command.strip().partition(" ") + cmd = cmd.lower() + arg = arg.strip() + + if cmd in ("/exit", "/quit", "/q"): + self.exit() + return + + if cmd == "/help": + help_text = Text("Available commands:\n", style="bold") + for hcmd, hdesc in _TUI_SLASH_COMMANDS: + help_text.append(f" {hcmd:<22}", style="cyan") + help_text.append(f"{hdesc}\n", style="dim") + self._mount_renderable(help_text) + return + + if cmd == "/current": + from .. import paths + + self._append_system(f"Thread: {self._thread_id}", style="dim") + if self._workspace_dir: + self._append_system( + f"Workspace: {_home_relative_path(self._workspace_dir)}", + style="dim", + ) + self._append_system("UI: textual", style="dim") + memory_path = _home_relative_path(str(paths.MEMORY_DIR)) + if memory_path: + self._append_system(f"Memory dir: {memory_path}", style="dim") + return + + if cmd == "/new": + # Clear all widgets except #welcome + container = self.query_one("#chat", VerticalScroll) + welcome = self.query_one("#welcome", Static) + for child in list(container.children): + if child is not welcome: + await child.remove() + + if not workspace_fixed: + self._workspace_dir = create_session_workspace(run_name) + self._thread_id = generate_thread_id() + self._agent = load_agent( + workspace_dir=self._workspace_dir, + checkpointer=self._checkpointer, + ) + if _channels_is_running(): + _ch_mod._cli_agent = self._agent + _ch_mod._cli_thread_id = self._thread_id + self._render_welcome() + self._render_status() + self._append_system(f"New session: {self._thread_id}", style="green") + return + + if cmd == "/clear": + container = self.query_one("#chat", VerticalScroll) + welcome = self.query_one("#welcome", Static) + for child in list(container.children): + if child is not welcome: + await child.remove() + return + + if cmd == "/threads": + await self._cmd_threads() + return + + if cmd == "/resume": + await self._cmd_resume(arg) + return + + if cmd == "/delete": + await self._cmd_delete(arg) + return + + if cmd == "/skills": + self._cmd_skills() + return + + if cmd == "/install-skill": + self._cmd_install_skill(arg) + return + + if cmd == "/uninstall-skill": + self._cmd_uninstall_skill(arg) + return + + if cmd == "/mcp": + self._cmd_mcp(arg) + return + + if cmd == "/channel": + self._cmd_channel(arg) + return + + self._append_system(f"Unknown command: {command}", style="yellow") + + async def _resolve_thread_id(self, prefix: str) -> str | None: + if await thread_exists(prefix): + return prefix + + similar = await find_similar_threads(prefix) + if len(similar) == 1: + return similar[0] + + if len(similar) > 1: + self._append_system( + f"Ambiguous thread ID '{prefix}'. Use a longer prefix.", + style="yellow", + ) + for thread in similar: + self._append_system(f" - {thread}", style="dim") + return None + + self._append_system(f"Thread '{prefix}' not found.", style="red") + return None + + async def _cmd_threads(self) -> None: + threads = await list_threads( + limit=0, + include_message_count=True, + include_preview=True, + ) + if not threads: + self._append_system("No saved sessions.", style="yellow") + return + + table = Table(title="Sessions", show_header=True, header_style="bold cyan") + table.add_column("ID", style="bold") + table.add_column("Preview", style="dim", max_width=50, no_wrap=True) + table.add_column("Messages", justify="right") + table.add_column("Model", style="dim") + table.add_column("Last Used", style="dim") + for thread in threads: + thread_id_value = thread["thread_id"] + marker = " *" if thread_id_value == self._thread_id else "" + table.add_row( + f"{thread_id_value}{marker}", + thread.get("preview", "") or "", + str(thread.get("message_count", 0)), + thread.get("model", "") or "", + _format_relative_time(thread.get("updated_at")), + ) + self._mount_renderable(table) + + async def _render_history(self, thread_id_value: str) -> None: + """Render conversation history from a saved thread.""" + messages = await get_thread_messages(thread_id_value) + if not messages: + return + + container = self.query_one("#chat", VerticalScroll) + await container.mount(SystemMessage("── Conversation history ──", msg_style="dim")) + for message in messages: + msg_type = getattr(message, "type", None) + content = getattr(message, "content", "") or "" + if isinstance(content, list): + parts = [ + block.get("text", "") + for block in content + if isinstance(block, dict) and block.get("type") == "text" + ] + content = " ".join(parts) if parts else "" + content = content.strip() + if len(content) > 220: + content = content[:220] + "..." + + if msg_type == "human": + await container.mount(UserMessage(content)) + elif msg_type == "ai": + tool_calls = getattr(message, "tool_calls", None) or [] + if content: + await container.mount( + Static(Text(content, style="dim")) + ) + if tool_calls: + names = [tc.get("name", "?") for tc in tool_calls] + await container.mount( + Static(Text(f" \u25b6 {', '.join(names)}", style="dim italic")) + ) + await container.mount(SystemMessage("── End of history ──", msg_style="dim")) + container.scroll_end(animate=False) + + async def _cmd_resume(self, arg: str) -> None: + if not arg: + self._append_system("Usage: /resume ", style="yellow") + await self._cmd_threads() + return + + resolved = await self._resolve_thread_id(arg) + if not resolved: + return + + metadata = await get_thread_metadata(resolved) + restored_workspace = (metadata or {}).get("workspace_dir", "") + if restored_workspace: + self._workspace_dir = restored_workspace + + self._thread_id = resolved + self._agent = load_agent( + workspace_dir=self._workspace_dir, + checkpointer=self._checkpointer, + ) + if _channels_is_running(): + _ch_mod._cli_agent = self._agent + _ch_mod._cli_thread_id = self._thread_id + self._render_welcome() + self._render_status() + self._append_system(f"Resumed session: {resolved}", style="green") + await self._render_history(resolved) + + async def _cmd_delete(self, arg: str) -> None: + if not arg: + self._append_system("Usage: /delete ", style="yellow") + return + + resolved = await self._resolve_thread_id(arg) + if not resolved: + return + + if resolved == self._thread_id: + self._append_system( + "Cannot delete the current session.", + style="yellow", + ) + return + + deleted = await delete_thread(resolved) + if deleted: + self._append_system(f"Deleted session {resolved}.", style="green") + else: + self._append_system(f"Session {resolved} not found.", style="red") + + def _cmd_skills(self) -> None: + from ..tools.skills_manager import list_skills + from ..paths import USER_SKILLS_DIR + + skills = list_skills(include_system=True) + if not skills: + self._append_system("No skills available.", style="dim") + self._append_system("Install with: /install-skill ", style="dim") + self._append_system(f"Skills directory: {_home_relative_path(str(USER_SKILLS_DIR))}", style="dim") + return + + user_skills = [s for s in skills if s.source == "user"] + system_skills = [s for s in skills if s.source == "system"] + + if user_skills: + table = Table(title=f"User Skills ({len(user_skills)})", show_header=True) + table.add_column("Name", style="green") + table.add_column("Description", style="dim") + for s in user_skills: + table.add_row(s.name, s.description) + self._mount_renderable(table) + + if system_skills: + table = Table(title=f"Built-in Skills ({len(system_skills)})", show_header=True) + table.add_column("Name", style="cyan") + table.add_column("Description", style="dim") + for s in system_skills: + table.add_row(s.name, s.description) + self._mount_renderable(table) + + self._append_system( + f"User skills folder: {_home_relative_path(str(USER_SKILLS_DIR))}", + style="dim", + ) + + def _cmd_install_skill(self, source: str) -> None: + from ..tools.skills_manager import install_skill + + if not source: + self._append_system("Usage: /install-skill ", style="yellow") + self._append_system("Examples:", style="dim") + self._append_system(" /install-skill ./my-skill", style="dim") + self._append_system( + " /install-skill https://github.com/user/repo/tree/main/skill-name", + style="dim", + ) + self._append_system(" /install-skill user/repo@skill-name", style="dim") + return + + self._append_system(f"Installing skill from: {source}", style="dim") + result = install_skill(source) + if result["success"]: + self._append_system(f"Installed: {result['name']}", style="green") + self._append_system( + f"Description: {result.get('description', '(none)')}", + style="dim", + ) + self._append_system(f"Path: {_home_relative_path(result['path'])}", style="dim") + self._append_system("Reload with /new to apply.", style="dim") + else: + self._append_system(f"Failed: {result['error']}", style="red") + + def _cmd_uninstall_skill(self, name: str) -> None: + from ..tools.skills_manager import uninstall_skill + + if not name: + self._append_system("Usage: /uninstall-skill ", style="yellow") + self._append_system("Use /skills to see installed skills.", style="dim") + return + + result = uninstall_skill(name) + if result["success"]: + self._append_system(f"Uninstalled: {name}", style="green") + self._append_system("Reload with /new to apply.", style="dim") + else: + self._append_system(f"Failed: {result['error']}", style="red") + + def _cmd_mcp(self, args: str) -> None: + args = args.strip() + if not args or args == "list": + self._mcp_list() + return + + parts = args.split(maxsplit=1) + subcmd = parts[0].lower() + subargs = parts[1] if len(parts) > 1 else "" + + if subcmd == "config": + self._mcp_config(subargs.strip()) + elif subcmd == "add": + self._mcp_add(subargs) + elif subcmd == "edit": + self._mcp_edit(subargs) + elif subcmd == "remove": + self._mcp_remove(subargs.strip()) + else: + self._append_system("MCP commands:", style="bold") + self._append_system(" /mcp List configured servers", style="dim") + self._append_system(" /mcp list List configured servers", style="dim") + self._append_system(" /mcp config Show detailed server config", style="dim") + self._append_system(" /mcp add ... Add a server", style="dim") + self._append_system(" /mcp edit ... Edit an existing server", style="dim") + self._append_system(" /mcp remove ... Remove a server", style="dim") + + def _mcp_list(self) -> None: + from ..mcp import load_mcp_config + from ..mcp.client import USER_MCP_CONFIG + + config = load_mcp_config() + if not config: + self._append_system("No MCP servers configured.", style="dim") + self._append_system("Add one with: /mcp add [args...]", style="dim") + return + + table = Table(title="MCP Servers", show_header=True) + table.add_column("Server", style="cyan") + table.add_column("Transport", style="green") + table.add_column("Tools", style="yellow") + table.add_column("Expose To", style="magenta") + + for name, server in config.items(): + transport = server.get("transport", "?") + tools = server.get("tools") + tools_str = ", ".join(tools) if tools else "(all)" + expose_to = server.get("expose_to", ["main"]) + if isinstance(expose_to, str): + expose_to = [expose_to] + expose_str = ", ".join(expose_to) + table.add_row(name, transport, tools_str, expose_str) + + self._mount_renderable(table) + self._append_system(f"Config file: {USER_MCP_CONFIG}", style="dim") + + def _mcp_config(self, name: str) -> None: + from ..mcp import load_mcp_config + from ..mcp.client import USER_MCP_CONFIG + + config = load_mcp_config() + if not config: + self._append_system("No MCP servers configured.", style="dim") + return + + if name and name not in config: + self._append_system(f"Server not found: {name}", style="red") + return + + servers = {name: config[name]} if name else config + for srv_name, srv in servers.items(): + table = Table(title=f"MCP Server: {srv_name}", show_header=True, title_style="bold cyan") + table.add_column("Setting", style="cyan") + table.add_column("Value") + table.add_row("transport", str(srv.get("transport", "(not set)"))) + if srv.get("command"): + table.add_row("command", str(srv["command"])) + if srv.get("args"): + table.add_row("args", " ".join(str(a) for a in srv["args"])) + if srv.get("url"): + table.add_row("url", str(srv["url"])) + if srv.get("headers"): + for k, v in srv["headers"].items(): + table.add_row(f"header: {k}", str(v)) + if srv.get("env"): + for k, v in srv["env"].items(): + table.add_row(f"env: {k}", str(v)) + tools = srv.get("tools") + table.add_row("tools", ", ".join(tools) if tools else "(all)") + expose_to = srv.get("expose_to", ["main"]) + if isinstance(expose_to, str): + expose_to = [expose_to] + table.add_row("expose_to", ", ".join(expose_to)) + self._mount_renderable(table) + + self._append_system(f"Config file: {USER_MCP_CONFIG}", style="dim") + + def _mcp_add(self, args_str: str) -> None: + from ..mcp import parse_mcp_add_args, add_mcp_server + + if not args_str.strip(): + self._append_system("Usage: /mcp add [args...]", style="yellow") + return + + try: + tokens = shlex.split(args_str) + kwargs = parse_mcp_add_args(tokens) + entry = add_mcp_server(**kwargs) + self._append_system( + f"Added MCP server: {kwargs['name']} ({entry['transport']})", + style="green", + ) + self._append_system("Reload with /new to apply.", style="dim") + except ValueError as exc: + self._append_system(f"{exc}", style="red") + + def _mcp_edit(self, args_str: str) -> None: + from ..mcp import parse_mcp_edit_args, edit_mcp_server + + if not args_str.strip(): + self._append_system( + "Usage: /mcp edit -- ...", + style="yellow", + ) + return + + try: + tokens = shlex.split(args_str) + name, fields = parse_mcp_edit_args(tokens) + if not fields: + self._append_system( + "No fields to edit. Use --transport, --command, --url, --tools, --expose-to, etc.", + style="red", + ) + return + edit_mcp_server(name, **fields) + self._append_system(f"Updated MCP server: {name}", style="green") + for k, v in fields.items(): + self._append_system(f" {k}: {v}", style="dim") + self._append_system("Reload with /new to apply.", style="dim") + except (KeyError, ValueError) as exc: + self._append_system(f"{exc}", style="red") + + def _mcp_remove(self, name: str) -> None: + from ..mcp import remove_mcp_server + + if not name: + self._append_system("Usage: /mcp remove ", style="yellow") + return + + if remove_mcp_server(name): + self._append_system(f"Removed MCP server: {name}", style="green") + self._append_system("Reload with /new to apply.", style="dim") + else: + self._append_system(f"Server not found: {name}", style="red") + + def _cmd_channel(self, args: str) -> None: + """Handle /channel command — start, stop, or show status.""" + from ..config import load_config + from .channel import ( + _add_channel_to_running_bus, + _start_channels_bus_mode, + _channels_running_list, + ) + + args = args.strip().lower() if args else "" + + if args == "status" or (not args and _channels_is_running()): + running = _channels_running_list() + if running and _ch_mod._manager: + detailed = _ch_mod._manager.get_detailed_status() + table = Table( + title="Channel Status", show_header=True, expand=False, + ) + table.add_column("Channel", style="cyan") + table.add_column("Status") + table.add_column("Uptime", style="dim") + table.add_column("Rx", justify="right") + table.add_column("Tx", justify="right") + for ch_name in running: + info = detailed.get(ch_name, {}) + secs = info.get("uptime_seconds", 0) + mins, s = divmod(int(secs), 60) + hours, mins = divmod(mins, 60) + uptime = ( + f"{hours}h{mins:02d}m" if hours else f"{mins}m{s:02d}s" + ) + rx = str(info.get("received", 0)) + tx = str(info.get("sent", 0)) + table.add_row( + ch_name, "[green]running[/green]", uptime, rx, tx, + ) + self._mount_renderable(table) + else: + self._append_system("No channel running", style="dim") + return + + if args.startswith("stop"): + stop_type = args[len("stop"):].strip() or None + if not _channels_is_running(): + self._append_system("No channel running", style="dim") + return + if stop_type: + if not _channels_is_running(stop_type): + self._append_system( + f"{stop_type} is not running", style="dim", + ) + return + _channels_stop(stop_type) + if stop_type in self._started_channel_types: + self._started_channel_types.remove(stop_type) + self._append_system(f"{stop_type} stopped", style="dim") + else: + running = _channels_running_list() + _channels_stop() + self._started_channel_types.clear() + self._append_system( + f"{', '.join(running)} stopped", style="dim", + ) + self._render_welcome() + return + + # Start channel(s) + app_config = load_config() + channel_type = args if args else ( + app_config.channel_enabled if app_config else "" + ) + if not channel_type: + self._append_system("No channel configured.", style="yellow") + self._append_system( + "Run EvoSci onboard or specify: /channel telegram", + style="dim", + ) + return + + requested = [ + t.strip() for t in channel_type.split(",") if t.strip() + ] + + if _channels_is_running(): + running = _channels_running_list() + results: list[tuple[str, bool, str]] = [] + for ct in requested: + if ct in running: + results.append((ct, True, "already running")) + else: + try: + _add_channel_to_running_bus( + ct, + app_config, + send_thinking=self._channel_send_thinking, + ) + results.append((ct, True, "connected (bus)")) + except Exception as e: + results.append((ct, False, str(e))) + else: + _ch_mod._cli_agent = self._agent + _ch_mod._cli_thread_id = self._thread_id + original = app_config.channel_enabled + app_config.channel_enabled = channel_type + try: + _start_channels_bus_mode( + app_config, + self._agent, + self._thread_id, + send_thinking=self._channel_send_thinking, + ) + results = [ + (ct, True, "connected (bus)") for ct in requested + ] + except Exception as e: + results = [(ct, False, str(e)) for ct in requested] + finally: + app_config.channel_enabled = original + + for ct, ok, _ in results: + if ok and ct not in self._started_channel_types: + self._started_channel_types.append(ct) + + self._render_channel_results(results) + self._render_welcome() + + def _render_channel_results( + self, results: list[tuple[str, bool, str]], + ) -> None: + for name, ok, detail in results: + if ok: + self._append_system( + f"\u25cf {name} {detail}", style="green", + ) + else: + self._append_system( + f"\u2717 {name} {detail}", style="yellow", + ) + + # ── Quit handling ────────────────────────────────────── + + def action_request_quit(self) -> None: + if self._busy: + if self._run_task is not None and not self._run_task.done(): + self._run_task.cancel() + self._append_system("Interrupted.", style="yellow") + else: + self._busy = False + prompt = self.query_one("#prompt", Input) + prompt.disabled = False + prompt.focus() + self._render_status() + self._append_system("Interrupted.", style="yellow") + return + # Clean up channels + if self._channel_timer is not None: + self._channel_timer.stop() + self._channel_timer = None + self._started_channel_types.clear() + if _channels_is_running(): + try: + _channels_stop() + except Exception: + pass + self.exit() + + # ── Banner & status ──────────────────────────────────── + + def _render_welcome(self) -> None: + channels_info: list[tuple[str, bool, str]] | None = None + try: + running = _channels_running_list() + started = self._started_channel_types + if running or started: + all_types = list(dict.fromkeys(running + started)) + channels_info = [ + (ct, True, "connected (bus)") for ct in all_types + ] + else: + from ..config import load_config + + cfg = load_config() + if cfg and cfg.channel_enabled: + types = [ + t.strip() + for t in cfg.channel_enabled.split(",") + if t.strip() + ] + if types: + channels_info = [ + (ct, False, "configured") for ct in types + ] + except Exception: + pass + + welcome = self.query_one("#welcome", Static) + welcome.update( + _build_welcome_banner( + thread_id=self._thread_id, + workspace_dir=self._workspace_dir, + mode=mode, + model=model, + provider=provider, + ui_backend="textual", + channels=channels_info, + ) + ) + + def _render_status(self) -> None: + status = self.query_one("#status", Static) + if self._busy: + left = "thinking..." + left_style = "bold #f59e0b" + else: + left = "manual | /help for commands" + left_style = "#f59e0b" + + model_label = model or "default-model" + if provider and model: + model_label = f"{provider}:{model}" + + status.update( + Text.assemble( + (left, left_style), + (" ", ""), + (model_label, "dim"), + ) + ) + + # ── Media forwarding helper (module-level) ────────────── + + _MEDIA_EXTENSIONS = { + ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".webp", ".svg", + ".pdf", ".mp3", ".wav", ".mp4", + } + + def _forward_media_to_channel( + state: StreamState, + tool_name: str, + media_sent: set[str], + send_fn: Any, + ) -> None: + """Check tool calls for media files and forward to channel.""" + import os + from ..paths import resolve_virtual_path + + arg_key = "path" if tool_name == "write_file" else "file_path" + for tc in reversed(state.tool_calls): + if tc.get("name") == tool_name: + p = tc.get("args", {}).get(arg_key, "") + if not p: + p = tc.get("args", {}).get("path", "") + if p and p not in media_sent: + ext = os.path.splitext(p)[1].lower() + if ext in _MEDIA_EXTENSIONS: + real_path = str(resolve_virtual_path(p)) + if not os.path.isfile(real_path) and os.path.isfile(p): + real_path = p + if os.path.isfile(real_path): + media_sent.add(p) + send_fn(real_path) + break + + # ── Entry point ───────────────────────────────────────── + + async def _amain() -> None: + async with get_checkpointer() as checkpointer: + effective_workspace = workspace_dir + effective_thread_id = thread_id + resumed = False + if thread_id: + if await thread_exists(thread_id): + resolved = thread_id + else: + similar = await find_similar_threads(thread_id) + resolved = similar[0] if len(similar) == 1 else None + if resolved: + meta = await get_thread_metadata(resolved) + ws = (meta or {}).get("workspace_dir", "") + if ws: + effective_workspace = ws + effective_thread_id = resolved + resumed = True + if not effective_thread_id: + effective_thread_id = generate_thread_id() + + initial_agent = load_agent( + workspace_dir=effective_workspace, + checkpointer=checkpointer, + ) + app = EvoTextualInteractiveApp( + agent=initial_agent, + thread_id_value=effective_thread_id, + workspace=effective_workspace, + checkpointer=checkpointer, + channel_send_thinking_value=channel_send_thinking, + resumed=resumed, + ) + await app.run_async() + + import nest_asyncio # type: ignore[import-untyped] + + nest_asyncio.apply() + try: + loop = asyncio.get_event_loop() + except RuntimeError: + loop = asyncio.new_event_loop() + asyncio.set_event_loop(loop) + loop.run_until_complete(_amain()) diff --git a/EvoScientist/cli/tui_runtime.py b/EvoScientist/cli/tui_runtime.py new file mode 100644 index 0000000..e737437 --- /dev/null +++ b/EvoScientist/cli/tui_runtime.py @@ -0,0 +1,99 @@ +"""Runtime selection for streaming UI backends.""" + +from __future__ import annotations + +from typing import Any, Callable + +from ..stream.display import console +from .tui_backends import RichStreamingBackend, StreamingTUIBackend + +DEFAULT_UI_BACKEND = "rich" +SUPPORTED_UI_BACKENDS = ("rich", "textual") + + +def normalize_ui_backend(value: str | None) -> str: + """Normalize user-provided backend name with a safe default.""" + if not value: + return DEFAULT_UI_BACKEND + normalized = value.strip().lower() + if normalized in SUPPORTED_UI_BACKENDS: + return normalized + return DEFAULT_UI_BACKEND + + +def _has_textual_support() -> bool: + try: + import textual # noqa: F401 + return True + except Exception: + return False + + +def resolve_ui_backend(value: str | None, *, warn_fallback: bool = False) -> str: + """Resolve requested backend and fallback safely when unavailable.""" + requested = normalize_ui_backend(value) + if requested == "textual" and not _has_textual_support(): + if warn_fallback: + console.print( + '[yellow]Textual TUI is unavailable (missing textual package). ' + 'Falling back to Rich.[/yellow]' + ) + return DEFAULT_UI_BACKEND + return requested + + +def get_backend(name: str | None, *, warn_fallback: bool = False) -> StreamingTUIBackend: + """Instantiate a streaming backend by name. + + Note: The Textual TUI is now a full interactive app (tui_interactive.py), + not a streaming backend. The streaming backend is always Rich. + """ + resolve_ui_backend(name, warn_fallback=warn_fallback) + return RichStreamingBackend() + + +def run_streaming( + *, + ui_backend: str | None, + agent: Any, + message: str, + thread_id: str, + show_thinking: bool, + interactive: bool, + on_thinking: Callable[[str], None] | None = None, + on_todo: Callable[[list[dict]], None] | None = None, + on_file_write: Callable[[str], None] | None = None, + metadata: dict | None = None, +) -> str: + """Run streaming with the selected backend.""" + backend = get_backend(ui_backend, warn_fallback=True) + try: + return backend.run_streaming( + agent=agent, + message=message, + thread_id=thread_id, + show_thinking=show_thinking, + interactive=interactive, + on_thinking=on_thinking, + on_todo=on_todo, + on_file_write=on_file_write, + metadata=metadata, + ) + except RuntimeError: + requested = normalize_ui_backend(ui_backend) + if requested == "textual": + console.print( + "[yellow]Textual TUI failed at runtime. Falling back to Rich for this request.[/yellow]" + ) + return RichStreamingBackend().run_streaming( + agent=agent, + message=message, + thread_id=thread_id, + show_thinking=show_thinking, + interactive=interactive, + on_thinking=on_thinking, + on_todo=on_todo, + on_file_write=on_file_write, + metadata=metadata, + ) + raise diff --git a/EvoScientist/cli/widgets/__init__.py b/EvoScientist/cli/widgets/__init__.py new file mode 100644 index 0000000..487d48f --- /dev/null +++ b/EvoScientist/cli/widgets/__init__.py @@ -0,0 +1,21 @@ +"""TUI widgets for EvoScientist Textual interface.""" + +from .loading_widget import LoadingWidget +from .thinking_widget import ThinkingWidget +from .assistant_message import AssistantMessage +from .tool_call_widget import ToolCallWidget +from .subagent_widget import SubAgentWidget +from .todo_widget import TodoWidget +from .user_message import UserMessage +from .system_message import SystemMessage + +__all__ = [ + "LoadingWidget", + "ThinkingWidget", + "AssistantMessage", + "ToolCallWidget", + "SubAgentWidget", + "TodoWidget", + "UserMessage", + "SystemMessage", +] diff --git a/EvoScientist/cli/widgets/assistant_message.py b/EvoScientist/cli/widgets/assistant_message.py new file mode 100644 index 0000000..9b9f7a4 --- /dev/null +++ b/EvoScientist/cli/widgets/assistant_message.py @@ -0,0 +1,50 @@ +"""Assistant message widget with incremental Markdown rendering.""" + +from __future__ import annotations + +from textual.containers import Vertical +from textual.widgets import Markdown + + +class AssistantMessage(Vertical): + """Displays the assistant's final Markdown response. + + Mount once, then call :meth:`append_content` for each text chunk. + When streaming finishes, call :meth:`stop_stream`. + + Each ``append_content`` call re-renders only *this* widget's Markdown — + not the entire chat history — which is the core improvement over the + old "rebuild Rich Group every 100 ms" approach. + """ + + DEFAULT_CSS = """ + AssistantMessage { + height: auto; + margin: 1 0 0 0; + } + AssistantMessage Markdown { + margin: 0; + padding: 0; + } + """ + + def __init__(self, initial_content: str = "") -> None: + super().__init__() + self._content = initial_content + + def compose(self): + yield Markdown("") + + def on_mount(self) -> None: + if self._content: + self.query_one(Markdown).update(self._content) + + async def append_content(self, text: str) -> None: + """Append text and re-render the Markdown widget.""" + self._content += text + self.query_one(Markdown).update(self._content) + + async def stop_stream(self) -> None: + """Finalize the stream — ensure final content is rendered.""" + if self._content: + self.query_one(Markdown).update(self._content) diff --git a/EvoScientist/cli/widgets/loading_widget.py b/EvoScientist/cli/widgets/loading_widget.py new file mode 100644 index 0000000..1b7f2dc --- /dev/null +++ b/EvoScientist/cli/widgets/loading_widget.py @@ -0,0 +1,50 @@ +"""Loading spinner widget shown while waiting for the first token.""" + +from __future__ import annotations + +from textual.widgets import Static + +_SPINNER_FRAMES = "\u280b\u2819\u2839\u2838\u283c\u2834\u2826\u2827\u2807\u280f" + + +class LoadingWidget(Static): + """Spinner + 'Thinking...' with elapsed time counter. + + Mount when a turn starts; call ``remove()`` when the first + thinking/text/tool_call event arrives. + """ + + DEFAULT_CSS = """ + LoadingWidget { + height: auto; + color: #22d3ee; + padding: 0 0; + } + """ + + def __init__(self) -> None: + super().__init__("") + self._frame = 0 + self._elapsed = 0.0 + self._timer_handle = None + + def on_mount(self) -> None: + self._timer_handle = self.set_interval(0.1, self._tick) + self._refresh_display() + + def _tick(self) -> None: + self._frame = (self._frame + 1) % len(_SPINNER_FRAMES) + self._elapsed += 0.1 + self._refresh_display() + + def _refresh_display(self) -> None: + char = _SPINNER_FRAMES[self._frame] + secs = int(self._elapsed) + self.update(f"{char} Thinking... ({secs}s)") + + async def cleanup(self) -> None: + """Stop timer and remove from DOM.""" + if self._timer_handle is not None: + self._timer_handle.stop() + self._timer_handle = None + await self.remove() diff --git a/EvoScientist/cli/widgets/subagent_widget.py b/EvoScientist/cli/widgets/subagent_widget.py new file mode 100644 index 0000000..8b33cf1 --- /dev/null +++ b/EvoScientist/cli/widgets/subagent_widget.py @@ -0,0 +1,186 @@ +"""Sub-agent widget — bordered area with nested tool calls.""" + +from __future__ import annotations + +from rich.text import Text + +from textual.containers import Vertical +from textual.widgets import Static + +from .tool_call_widget import ToolCallWidget + +_SPINNER_FRAMES = "\u280b\u2819\u2839\u2838\u283c\u2834\u2826\u2827\u2807\u280f" + + +class SubAgentWidget(Vertical): + """Displays a sub-agent's activity with a bordered frame. + + Active state:: + + ┌ ▶ Cooking with research-agent — Search literature ─┐ + │ ● tavily_search query="LLM attention" │ + │ ✓ 3 results │ + └─────────────────────────────────────────────────────┘ + + Completed state:: + + ✓ Cooking with research-agent (3 tools) + """ + + DEFAULT_CSS = """ + SubAgentWidget { + height: auto; + margin: 0 0; + } + SubAgentWidget .sa-header { + height: auto; + color: #22d3ee; + } + SubAgentWidget .sa-tools { + height: auto; + padding: 0 0 0 2; + } + SubAgentWidget .sa-footer { + height: auto; + color: #22d3ee; + } + SubAgentWidget.--completed .sa-header { + color: #4ade80; + } + SubAgentWidget.--completed .sa-footer { + color: #4ade80; + } + """ + + def __init__(self, name: str, description: str = "") -> None: + super().__init__() + self._sa_name = name + self._description = description + self._is_active = True + self._frame = 0 + self._tool_count = 0 + self._timer_handle = None + self._tool_widgets: dict[str, ToolCallWidget] = {} + + @property + def sa_name(self) -> str: + return self._sa_name + + def update_name(self, name: str, description: str = "") -> None: + """Update the sub-agent display name after resolution.""" + self._sa_name = name + if description: + self._description = description + try: + self._render_header() + except Exception: + pass # Widget may not be mounted yet + + def compose(self): + yield Static("", classes="sa-header") + yield Vertical(classes="sa-tools") + yield Static("", classes="sa-footer") + + def on_mount(self) -> None: + self._timer_handle = self.set_interval(0.1, self._tick) + self._render_header() + self._render_footer() + + def _tick(self) -> None: + if self._is_active: + self._frame = (self._frame + 1) % len(_SPINNER_FRAMES) + self._render_header() + + def _display_name(self) -> str: + name = f"Cooking with {self._sa_name}" + if self._description: + desc = self._description.split("\n")[0].strip() + if len(desc) > 50: + desc = desc[:47] + "\u2026" + name += f" \u2014 {desc}" + return name + + def _render_header(self) -> None: + header = self.query_one(".sa-header", Static) + line = Text() + if self._is_active: + char = _SPINNER_FRAMES[self._frame] + line.append(f"\u250c \u25b6 {self._display_name()} {char}", style="bold cyan") + else: + line.append(f"\u2713 {self._display_name()}", style="bold green") + line.append(f" ({self._tool_count} tools)", style="dim") + header.update(line) + + def _render_footer(self) -> None: + footer = self.query_one(".sa-footer", Static) + if self._is_active: + footer.update(Text("\u2514 running...", style="dim cyan")) + else: + footer.update(Text("")) + + async def add_tool_call( + self, + tool_name: str, + tool_args: dict | None = None, + tool_id: str = "", + ) -> ToolCallWidget: + """Mount a new ToolCallWidget inside this sub-agent.""" + self._tool_count += 1 + w = ToolCallWidget(tool_name, tool_args, tool_id) + tools_container = self.query_one(".sa-tools", Vertical) + await tools_container.mount(w) + if tool_id: + self._tool_widgets[tool_id] = w + return w + + def complete_tool( + self, + tool_name: str, + content: str, + success: bool = True, + tool_id: str = "", + ) -> None: + """Update the matching ToolCallWidget with its result.""" + widget = None + if tool_id and tool_id in self._tool_widgets: + widget = self._tool_widgets[tool_id] + else: + # Match by name — find first running tool with this name + for w in self._tool_widgets.values(): + if w.tool_name == tool_name and w._status == "running": + widget = w + break + if widget is None: + # Fallback: find any running tool + tools = self.query_one(".sa-tools", Vertical) + for child in tools.children: + if isinstance(child, ToolCallWidget) and child._status == "running": + if child.tool_name == tool_name: + widget = child + break + + if widget is not None: + if success: + widget.set_success(content) + else: + widget.set_error(content) + + def finalize(self) -> None: + """Mark sub-agent as completed and stop all nested timers.""" + self._is_active = False + if self._timer_handle is not None: + self._timer_handle.stop() + self._timer_handle = None + # Stop timers on any nested ToolCallWidgets still running + for tw in self._tool_widgets.values(): + if tw._status == "running": + tw._stop_timer() + tw._status = "success" + try: + tw._render_header() + tw._render_status() + except Exception: + pass + self.add_class("--completed") + self._render_header() + self._render_footer() diff --git a/EvoScientist/cli/widgets/system_message.py b/EvoScientist/cli/widgets/system_message.py new file mode 100644 index 0000000..368e4dc --- /dev/null +++ b/EvoScientist/cli/widgets/system_message.py @@ -0,0 +1,20 @@ +"""System message widget.""" + +from __future__ import annotations + +from rich.text import Text + +from textual.widgets import Static + + +class SystemMessage(Static): + """Displays a system/status message (replaces ``_append_system``).""" + + DEFAULT_CSS = """ + SystemMessage { + height: auto; + } + """ + + def __init__(self, content: str, *, msg_style: str = "dim") -> None: + super().__init__(Text(content, style=msg_style)) diff --git a/EvoScientist/cli/widgets/thinking_widget.py b/EvoScientist/cli/widgets/thinking_widget.py new file mode 100644 index 0000000..a76eb36 --- /dev/null +++ b/EvoScientist/cli/widgets/thinking_widget.py @@ -0,0 +1,82 @@ +"""Thinking panel widget for extended thinking display. + +Renders a Rich Panel matching the Rich CLI's style: +blue border, "Thinking" title with spinner, dim content. +""" + +from __future__ import annotations + +from rich.panel import Panel +from rich.text import Text + +from textual.widgets import Static + +_SPINNER_FRAMES = "\u280b\u2819\u2839\u2838\u283c\u2834\u2826\u2827\u2807\u280f" +_MAX_DISPLAY_CHARS = 1000 + + +class ThinkingWidget(Static): + """Collapsible panel showing the model's extended thinking. + + Uses Rich ``Panel`` for rendering, matching the Rich CLI output exactly. + + Usage:: + + w = ThinkingWidget(show_thinking=True) + await container.mount(w) + w.append_text("reasoning chunk...") + w.finalize() # stop spinner + """ + + DEFAULT_CSS = """ + ThinkingWidget { + height: auto; + margin: 0 0 1 0; + } + """ + + def __init__(self, *, show_thinking: bool = True) -> None: + super().__init__("") + self._content = "" + self._is_active = True + self._show = show_thinking + self._frame = 0 + self._timer_handle = None + if not show_thinking: + self.display = False + + def on_mount(self) -> None: + self._timer_handle = self.set_interval(0.1, self._tick) + self._refresh_display() + + def _tick(self) -> None: + if self._is_active: + self._frame = (self._frame + 1) % len(_SPINNER_FRAMES) + self._refresh_display() + + def _refresh_display(self) -> None: + if self._is_active: + char = _SPINNER_FRAMES[self._frame] + title = f"Thinking {char}" + else: + title = "Thinking" + + display = self._content.rstrip() + if len(display) > _MAX_DISPLAY_CHARS: + display = "..." + display[-_MAX_DISPLAY_CHARS:] + + body = Text(display, style="dim") if display else Text("...", style="dim") + self.update(Panel(body, title=title, border_style="blue", padding=(0, 1))) + + def append_text(self, chunk: str) -> None: + """Append a chunk of thinking text.""" + self._content += chunk + self._refresh_display() + + def finalize(self) -> None: + """Mark thinking as complete — stop spinner, dim border.""" + self._is_active = False + if self._timer_handle is not None: + self._timer_handle.stop() + self._timer_handle = None + self._refresh_display() diff --git a/EvoScientist/cli/widgets/todo_widget.py b/EvoScientist/cli/widgets/todo_widget.py new file mode 100644 index 0000000..029c55e --- /dev/null +++ b/EvoScientist/cli/widgets/todo_widget.py @@ -0,0 +1,77 @@ +"""Todo/Task list panel widget. + +Renders a Rich Panel matching the Rich CLI's ``_render_todo_panel()`` style: +cyan border, centered "Task List" title, status icons per item. +""" + +from __future__ import annotations + +from rich.panel import Panel +from rich.text import Text + +from textual.widgets import Static + + +class TodoWidget(Static): + """Displays a task list panel matching the Rich CLI style. + + Usage:: + + w = TodoWidget(items) + await container.mount(w) + w.update_items(new_items) # re-render in place + """ + + DEFAULT_CSS = """ + TodoWidget { + height: auto; + margin: 1 0; + } + """ + + def __init__(self, items: list[dict] | None = None) -> None: + super().__init__("") + self._items = items or [] + + def on_mount(self) -> None: + self._refresh_display() + + def update_items(self, items: list[dict]) -> None: + """Replace the task list and re-render.""" + self._items = items + self._refresh_display() + + def _refresh_display(self) -> None: + if not self._items: + self.update(Text("No tasks", style="dim")) + return + + lines = Text() + for i, item in enumerate(self._items): + if i > 0: + lines.append("\n") + status = str(item.get("status", "todo")).lower() + content = str( + item.get("content", item.get("task", item.get("title", ""))) + ) + + if status in ("done", "completed", "complete"): + symbol = "\u2713" + style = "green dim" + elif status in ("active", "in_progress", "in-progress", "working"): + symbol = "\u23f3" + style = "yellow" + else: + symbol = "\u25a1" + style = "dim" + + lines.append(f"{symbol} ", style=style) + lines.append(content, style=style) + + self.update(Panel( + lines, + title="Task List", + title_align="center", + border_style="cyan", + padding=(0, 1), + )) diff --git a/EvoScientist/cli/widgets/tool_call_widget.py b/EvoScientist/cli/widgets/tool_call_widget.py new file mode 100644 index 0000000..c913e94 --- /dev/null +++ b/EvoScientist/cli/widgets/tool_call_widget.py @@ -0,0 +1,174 @@ +"""Tool call widget with status lifecycle and collapsible output.""" + +from __future__ import annotations + +from rich.text import Text + +from textual.containers import Vertical +from textual.widgets import Static + +from ...stream.utils import format_tool_compact + +_SPINNER_FRAMES = "\u280b\u2819\u2839\u2838\u283c\u2834\u2826\u2827\u2807\u280f" + +# Output is collapsed when it exceeds these limits +_COLLAPSE_LINES = 6 +_COLLAPSE_CHARS = 400 + + +class ToolCallWidget(Vertical): + """Displays a single tool call with running spinner → result. + + Lifecycle: ``running`` → ``success`` | ``error`` + + Usage:: + + w = ToolCallWidget("read_file", {"path": "/foo.py"}, tool_id="abc") + await container.mount(w) + # ... later ... + w.set_success("[OK] 42 lines") + """ + + DEFAULT_CSS = """ + ToolCallWidget { + height: auto; + padding: 0 0; + } + ToolCallWidget .tool-header { + height: auto; + } + ToolCallWidget .tool-status { + height: auto; + padding: 0 0 0 2; + } + ToolCallWidget .tool-output { + height: auto; + padding: 0 0 0 4; + color: #9ca3af; + display: none; + } + ToolCallWidget .tool-output.--visible { + display: block; + } + """ + + def __init__( + self, + tool_name: str, + tool_args: dict | None = None, + tool_id: str = "", + ) -> None: + super().__init__() + self._tool_name = tool_name + self._tool_args = tool_args or {} + self._tool_id = tool_id + self._status = "running" + self._result_content = "" + self._frame = 0 + self._elapsed = 0.0 + self._timer_handle = None + self._collapsed = True + + @property + def tool_id(self) -> str: + return self._tool_id + + @property + def tool_name(self) -> str: + return self._tool_name + + def compose(self): + yield Static("", classes="tool-header") + yield Static("", classes="tool-status") + yield Static("", classes="tool-output") + + def on_mount(self) -> None: + self._timer_handle = self.set_interval(0.1, self._tick) + self._render_header() + self._render_status() + + def _tick(self) -> None: + if self._status == "running": + self._frame = (self._frame + 1) % len(_SPINNER_FRAMES) + self._elapsed += 0.1 + self._render_status() + + def _render_header(self) -> None: + compact = format_tool_compact(self._tool_name, self._tool_args) + header = self.query_one(".tool-header", Static) + line = Text() + if self._status == "running": + line.append("\u25cf ", style="bold yellow") + line.append(compact, style="bold yellow") + elif self._status == "success": + line.append("\u2713 ", style="bold green") + line.append(compact, style="bold green") + else: + line.append("\u2717 ", style="bold red") + line.append(compact, style="bold red") + header.update(line) + + def _render_status(self) -> None: + status_w = self.query_one(".tool-status", Static) + if self._status == "running": + char = _SPINNER_FRAMES[self._frame] + secs = int(self._elapsed) + status_w.update(Text(f"{char} Running... ({secs}s)", style="yellow dim")) + elif self._status == "success": + # Show first line of result as summary + summary = self._result_summary() + status_w.update(Text(f"\u2713 {summary}", style="green dim")) + else: + summary = self._result_summary() + status_w.update(Text(f"\u2717 {summary}", style="red dim")) + + def _result_summary(self) -> str: + """One-line summary of the result.""" + if not self._result_content: + return "done" + first_line = self._result_content.strip().split("\n")[0] + if len(first_line) > 60: + first_line = first_line[:57] + "\u2026" + return first_line + + def _should_collapse(self) -> bool: + lines = self._result_content.strip().split("\n") + return len(lines) > _COLLAPSE_LINES or len(self._result_content) > _COLLAPSE_CHARS + + def set_success(self, content: str) -> None: + """Mark tool call as successfully completed.""" + self._status = "success" + self._result_content = content + self._stop_timer() + self._render_header() + self._render_status() + self._render_output() + + def set_error(self, content: str) -> None: + """Mark tool call as failed.""" + self._status = "error" + self._result_content = content + self._stop_timer() + self._render_header() + self._render_status() + self._render_output() + + def _render_output(self) -> None: + output_w = self.query_one(".tool-output", Static) + if not self._result_content.strip(): + return + if self._status == "error" or not self._should_collapse(): + # Show full output for errors or short output + style = "red dim" if self._status == "error" else "dim" + content = self._result_content.strip() + if len(content) > 800: + content = content[:800] + "\n... (truncated)" + output_w.update(Text(content, style=style)) + output_w.add_class("--visible") + # Otherwise collapsed — user can click to expand + # (click handling left for future enhancement) + + def _stop_timer(self) -> None: + if self._timer_handle is not None: + self._timer_handle.stop() + self._timer_handle = None diff --git a/EvoScientist/cli/widgets/user_message.py b/EvoScientist/cli/widgets/user_message.py new file mode 100644 index 0000000..21e3809 --- /dev/null +++ b/EvoScientist/cli/widgets/user_message.py @@ -0,0 +1,25 @@ +"""User message widget.""" + +from __future__ import annotations + +from rich.text import Text + +from textual.widgets import Static + + +class UserMessage(Static): + """Displays user input with a blue prompt marker.""" + + DEFAULT_CSS = """ + UserMessage { + height: auto; + margin: 1 0 0 0; + } + """ + + def __init__(self, content: str) -> None: + renderable = Text.assemble( + ("> ", "bold #38bdf8"), + (content, "#e5e7eb"), + ) + super().__init__(renderable) diff --git a/EvoScientist/config/onboard.py b/EvoScientist/config/onboard.py index b2542ed..cf1105e 100644 --- a/EvoScientist/config/onboard.py +++ b/EvoScientist/config/onboard.py @@ -90,7 +90,7 @@ def _checkbox_ask(choices, message: str, **kwargs): finally: InquirerControl._get_choice_tokens = original -STEPS = ["Provider", "API Key", "Model", "Tavily Key", "Workspace", "Parameters", "Skills", "MCP Servers", "Channels"] +STEPS = ["UI", "Provider", "API Key", "Model", "Tavily Key", "Workspace", "Parameters", "Skills", "MCP Servers", "Channels"] # ============================================================================= @@ -351,6 +351,35 @@ def _print_step_skipped(step_name: str, reason: str = "kept current") -> None: # Step Functions # ============================================================================= +def _step_ui_backend(config: EvoScientistConfig) -> str: + """Step 0: Select UI backend (Rich CLI or Textual TUI). + + Args: + config: Current configuration. + + Returns: + Selected backend name ("rich" or "textual"). + """ + choices = [ + Choice(title="Rich CLI (classic terminal)", value="rich"), + Choice(title="Textual TUI (full-screen interface)", value="textual"), + ] + + backend = questionary.select( + "Select UI mode:", + choices=choices, + default=config.ui_backend, + style=WIZARD_STYLE, + qmark=QMARK, + use_indicator=True, + ).ask() + + if backend is None: + raise KeyboardInterrupt() + + return backend + + def _step_provider(config: EvoScientistConfig) -> str: """Step 1: Select LLM provider. @@ -1661,6 +1690,10 @@ def run_onboard(skip_validation: bool = False) -> bool: # Load existing config as starting point config = load_config() + # Step 0: UI Backend + ui_backend = _step_ui_backend(config) + config.ui_backend = ui_backend + # Step 1: Provider provider = _step_provider(config) config.provider = provider diff --git a/EvoScientist/config/settings.py b/EvoScientist/config/settings.py index 6f80144..04ae92c 100644 --- a/EvoScientist/config/settings.py +++ b/EvoScientist/config/settings.py @@ -83,6 +83,7 @@ class EvoScientistConfig: # UI Settings show_thinking: bool = True + ui_backend: Literal["rich", "textual"] = "rich" # Channel Settings channel_enabled: str = "" # "imessage" | "telegram" | "discord" | "slack" | "wechat" | "dingtalk" | "feishu" | "email" | "qq" | "signal" | "" (comma-separated for multiple) @@ -331,6 +332,7 @@ _ENV_MAPPINGS = { "tavily_api_key": "TAVILY_API_KEY", "default_mode": "EVOSCIENTIST_DEFAULT_MODE", "default_workdir": "EVOSCIENTIST_WORKSPACE_DIR", + "ui_backend": "EVOSCIENTIST_UI_BACKEND", } diff --git a/EvoScientist/stream/display.py b/EvoScientist/stream/display.py index e334311..bda04ee 100644 --- a/EvoScientist/stream/display.py +++ b/EvoScientist/stream/display.py @@ -19,7 +19,7 @@ from rich.text import Text # type: ignore[import-untyped] from ..paths import resolve_virtual_path from .formatter import ToolResultFormatter -from .state import StreamState, SubAgentState, _build_todo_stats, _parse_todo_items +from .state import StreamState, SubAgentState, _build_todo_stats, _parse_todo_items, _INTERNAL_TOOLS from .utils import DisplayLimits, ToolStatus, format_tool_compact, is_success from .events import stream_agent_events @@ -376,6 +376,10 @@ def create_streaming_display( tr = tool_results[i] if has_result else None is_task = tc.get('name') == 'task' + # Skip internal middleware tools + if tc.get('name') in _INTERNAL_TOOLS: + continue + if is_task: # Skip task calls with empty args (still streaming) if tc.get('args'): @@ -423,10 +427,19 @@ def create_streaming_display( # Task tool calls are rendered as part of sub-agent sections below - # Response text handling - has_pending_tools = len(tool_calls) > len(tool_results) + # Response text handling — exclude internal tools (e.g. ExtractedMemory) + # from the "done" calculation so they don't block final Markdown rendering. + _n_visible = 0 + _n_visible_done = 0 + for i, tc in enumerate(tool_calls): + if tc.get('name') in _INTERNAL_TOOLS: + continue + _n_visible += 1 + if i < len(tool_results): + _n_visible_done += 1 + has_pending_tools = _n_visible > _n_visible_done any_active_subagent = any(sa.is_active for sa in subagents) - has_used_tools = len(tool_calls) > 0 + has_used_tools = _n_visible > 0 all_done = not has_pending_tools and not any_active_subagent and not is_processing # Intermediate narration (tools still running) -- dim italic above Task List @@ -459,12 +472,12 @@ def create_streaming_display( spinner = Spinner("dots", text=" Analyzing results...", style="cyan") elements.append(spinner) - # Final response -- render as Markdown when all work is done + # Final response -- render as streaming Markdown whenever all tools are done. + # The Live display is transient; display_final_results() re-renders the + # permanent output after Live exits, so no visible duplication occurs. if response_text and all_done: elements.append(Text("")) # blank separator elements.append(Markdown(response_text)) - elif is_responding and not thinking_text and not has_pending_tools: - elements.append(Text("Generating response...", style="dim")) return Group(*elements) if elements else Text("Processing...", style="dim") @@ -498,7 +511,12 @@ def display_final_results( has_result = i < len(state.tool_results) tr = state.tool_results[i] if has_result else None content = tr.get('content', '') if tr is not None else '' - is_task = tc.get('name', '').lower() == 'task' + tool_name = tc.get('name', '') + is_task = tool_name.lower() == 'task' + + # Skip internal middleware tools + if tool_name in _INTERNAL_TOOLS: + continue # Task tools: show delegation line + compact sub-agent summary if is_task: diff --git a/EvoScientist/stream/state.py b/EvoScientist/stream/state.py index f17492e..4d20128 100644 --- a/EvoScientist/stream/state.py +++ b/EvoScientist/stream/state.py @@ -7,6 +7,10 @@ No Rich dependencies — stdlib only. import ast import json +# Tool names that are internal middleware artifacts (not user-visible actions). +# These should be excluded from display rendering and "all_done" calculations. +_INTERNAL_TOOLS = {"ExtractedMemory"} + class SubAgentState: """Tracks a single sub-agent's activity.""" @@ -186,8 +190,9 @@ class StreamState: self.todo_items = todos elif event_type == "tool_result": - self.is_processing = True result_name = event.get("name", "unknown") + if result_name not in _INTERNAL_TOOLS: + self.is_processing = True result_content = event.get("content", "") self.tool_results.append({ "name": result_name, diff --git a/README.md b/README.md index a89078f..9bbd472 100644 --- a/README.md +++ b/README.md @@ -173,9 +173,14 @@ EvoSci # or EvoScientist --use-cwd Use current working directory as workspace --thread-id Resume a conversation thread --no-thinking Disable thinking display +--ui UI backend: rich (default) or textual (beta) -p, --prompt Single-shot mode: execute query and exit ``` +> [!NOTE] +> In `--ui textual` mode, the built-in TUI commands are: +> `/help`, `/current`, `/new`, `/clear`, `/threads`, `/resume `, `/delete `, `/exit`. + ![demo](./assets/EvoScientist_cli_help.png) **Configuration commands:** @@ -196,7 +201,7 @@ EvoSci config path # Show config file path |---------|-------------| | `/exit` | Quit the session | | `/new` | Start a new session (new workspace + thread) | -| `/thread` | Show current thread ID and workspace path | +| `/current` | Show current thread ID and workspace path | | `/channel` | Start iMessage channel (shares agent session) | | `/skills` | List installed user skills | | `/install-skill ` | Install a skill from local path or GitHub | diff --git a/pyproject.toml b/pyproject.toml index 5009360..4dfa941 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ dependencies = [ "markdownify>=0.14", "nest-asyncio>=1.6", "langchain-mcp-adapters>=0.1", + "textual>=0.80", ] [project.optional-dependencies] diff --git a/tests/test_channel_comprehensive.py b/tests/test_channel_comprehensive.py index 6b4949c..45bfd3c 100644 --- a/tests/test_channel_comprehensive.py +++ b/tests/test_channel_comprehensive.py @@ -1121,6 +1121,55 @@ class TestChannelManagerDrain: _run(_test()) +class TestChannelManagerTracking: + + def test_record_message(self): + bus = MessageBus() + mgr = ChannelManager(bus) + mgr.register(StubChannel()) + + mgr.record_message("stub", "received") + mgr.record_message("stub", "received") + mgr.record_message("stub", "sent") + + assert mgr._message_counts["stub"]["received"] == 2 + assert mgr._message_counts["stub"]["sent"] == 1 + + def test_record_message_unknown_channel(self): + bus = MessageBus() + mgr = ChannelManager(bus) + + # Should not raise, auto-creates entry + mgr.record_message("unknown", "received") + assert mgr._message_counts["unknown"]["received"] == 1 + + def test_get_detailed_status(self): + bus = MessageBus() + mgr = ChannelManager(bus) + mgr.register(StubChannel()) + + # Simulate start_all setting start_times + mgr._start_times["stub"] = datetime.now() + mgr._message_counts["stub"] = {"received": 5, "sent": 3} + + status = mgr.get_detailed_status() + assert "stub" in status + assert status["stub"]["registered"] is True + assert status["stub"]["received"] == 5 + assert status["stub"]["sent"] == 3 + assert status["stub"]["uptime_seconds"] >= 0 + assert status["stub"]["start_time"] is not None + + def test_get_detailed_status_no_start_time(self): + bus = MessageBus() + mgr = ChannelManager(bus) + mgr.register(StubChannel()) + + status = mgr.get_detailed_status() + assert status["stub"]["uptime_seconds"] == 0 + assert status["stub"]["start_time"] is None + + class TestChannelManagerStatus: def test_get_status(self): @@ -1365,6 +1414,12 @@ class TestMessageBus: # dispatch should survive the error _run(_test()) + def test_stop_flag(self): + bus = MessageBus() + assert bus._running is False + bus.stop() + assert bus._running is False + # ═══════════════════════════════════════════════════════════════════ # 9. Event dataclasses diff --git a/tests/test_channel_manager.py b/tests/test_channel_manager.py deleted file mode 100644 index 0b5740c..0000000 --- a/tests/test_channel_manager.py +++ /dev/null @@ -1,160 +0,0 @@ -"""Tests for ChannelManager.""" - -import asyncio - -import pytest - -from EvoScientist.channels.bus.message_bus import MessageBus -from EvoScientist.channels.channel_manager import ChannelManager -from EvoScientist.channels.base import Channel, OutboundMessage - - -from tests.conftest import run_async as _run - - -class _FakeConfig: - text_chunk_limit = 4096 - allowed_senders = None - - -class FakeChannel(Channel): - """Minimal channel for testing.""" - - name = "fake" - - def __init__(self): - super().__init__(_FakeConfig()) - self._started = False - self._stopped = False - self._sent: list[OutboundMessage] = [] - - async def start(self): - self._started = True - - async def stop(self): - self._stopped = True - - async def receive(self): - while True: - try: - msg = await asyncio.wait_for( - self._queue.get(), timeout=0.5, - ) - yield msg - except asyncio.TimeoutError: - return - - async def send(self, message: OutboundMessage) -> bool: - self._sent.append(message) - return True - - async def _send_chunk(self, chat_id, formatted_text, raw_text, reply_to, metadata): - pass - - -class TestChannelManagerRegister: - def test_register_channel(self): - bus = MessageBus() - mgr = ChannelManager(bus) - ch = FakeChannel() - result = mgr.register(ch) - assert "fake" in mgr.enabled_channels - assert mgr.get_channel("fake") is ch - assert result is ch - - def test_duplicate_register_raises(self): - bus = MessageBus() - mgr = ChannelManager(bus) - mgr.register(FakeChannel()) - with pytest.raises(ValueError, match="already registered"): - mgr.register(FakeChannel()) - - def test_get_status(self): - bus = MessageBus() - mgr = ChannelManager(bus) - mgr.register(FakeChannel()) - status = mgr.get_status() - assert "fake" in status - assert status["fake"]["registered"] is True - - -class TestChannelManagerDispatch: - def test_outbound_dispatch_routes_to_channel(self): - async def _test(): - bus = MessageBus() - mgr = ChannelManager(bus) - ch = FakeChannel() - mgr.register(ch) - - # Start only the dispatcher (not full start_all) - dispatch = asyncio.create_task( - mgr._dispatch_outbound() - ) - - # Publish an outbound message - await bus.publish_outbound(OutboundMessage( - channel="fake", chat_id="u1", - content="hello from agent", - )) - - await asyncio.sleep(0.1) - dispatch.cancel() - try: - await dispatch - except asyncio.CancelledError: - pass - - assert len(ch._sent) == 1 - assert ch._sent[0].content == "hello from agent" - assert ch._sent[0].chat_id == "u1" - - _run(_test()) - - -class TestChannelManagerTracking: - def test_record_message(self): - bus = MessageBus() - mgr = ChannelManager(bus) - mgr.register(FakeChannel()) - - mgr.record_message("fake", "received") - mgr.record_message("fake", "received") - mgr.record_message("fake", "sent") - - assert mgr._message_counts["fake"]["received"] == 2 - assert mgr._message_counts["fake"]["sent"] == 1 - - def test_record_message_unknown_channel(self): - bus = MessageBus() - mgr = ChannelManager(bus) - - # Should not raise, auto-creates entry - mgr.record_message("unknown", "received") - assert mgr._message_counts["unknown"]["received"] == 1 - - def test_get_detailed_status(self): - bus = MessageBus() - mgr = ChannelManager(bus) - mgr.register(FakeChannel()) - - # Simulate start_all setting start_times - from datetime import datetime - mgr._start_times["fake"] = datetime.now() - mgr._message_counts["fake"] = {"received": 5, "sent": 3} - - status = mgr.get_detailed_status() - assert "fake" in status - assert status["fake"]["registered"] is True - assert status["fake"]["received"] == 5 - assert status["fake"]["sent"] == 3 - assert status["fake"]["uptime_seconds"] >= 0 - assert status["fake"]["start_time"] is not None - - def test_get_detailed_status_no_start_time(self): - bus = MessageBus() - mgr = ChannelManager(bus) - mgr.register(FakeChannel()) - - status = mgr.get_detailed_status() - assert status["fake"]["uptime_seconds"] == 0 - assert status["fake"]["start_time"] is None diff --git a/tests/test_cli_tui_dispatch.py b/tests/test_cli_tui_dispatch.py new file mode 100644 index 0000000..d0edb08 --- /dev/null +++ b/tests/test_cli_tui_dispatch.py @@ -0,0 +1,53 @@ +"""Tests for CLI interactive UI backend dispatch.""" + +from EvoScientist.cli.interactive import cmd_interactive + + +def test_cmd_interactive_dispatches_to_textual(monkeypatch): + captured: dict[str, object] = {} + + def _fake_resolve_ui_backend(value, *, warn_fallback=False): # noqa: ANN001 + captured["resolved_input"] = value + captured["warn_fallback"] = warn_fallback + return "textual" + + def _fake_run_textual_interactive(**kwargs): # noqa: ANN003 + captured["kwargs"] = kwargs + + monkeypatch.setattr( + "EvoScientist.cli.interactive.resolve_ui_backend", + _fake_resolve_ui_backend, + ) + monkeypatch.setattr( + "EvoScientist.cli.interactive.run_textual_interactive", + _fake_run_textual_interactive, + ) + + cmd_interactive( + show_thinking=True, + channel_send_thinking=True, + workspace_dir="/tmp/workspace", + workspace_fixed=True, + mode="daemon", + model="demo-model", + provider="demo-provider", + run_name="demo-run", + thread_id="thread-1", + ui_backend="textual", + ) + + assert captured["resolved_input"] == "textual" + assert captured["warn_fallback"] is True + + kwargs = captured["kwargs"] + assert isinstance(kwargs, dict) + assert kwargs["workspace_dir"] == "/tmp/workspace" + assert kwargs["workspace_fixed"] is True + assert kwargs["mode"] == "daemon" + assert kwargs["model"] == "demo-model" + assert kwargs["provider"] == "demo-provider" + assert kwargs["run_name"] == "demo-run" + assert kwargs["thread_id"] == "thread-1" + assert kwargs["channel_send_thinking"] is True + assert callable(kwargs["load_agent"]) + assert callable(kwargs["create_session_workspace"]) diff --git a/tests/test_config.py b/tests/test_config.py index 3993acb..6c8ef86 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -38,6 +38,7 @@ def temp_config_dir(tmp_path, monkeypatch): "TAVILY_API_KEY", "EVOSCIENTIST_DEFAULT_MODE", "EVOSCIENTIST_WORKSPACE_DIR", + "EVOSCIENTIST_UI_BACKEND", ]: monkeypatch.delenv(key, raising=False) return config_dir @@ -52,6 +53,7 @@ def clean_env(monkeypatch): "TAVILY_API_KEY", "EVOSCIENTIST_DEFAULT_MODE", "EVOSCIENTIST_WORKSPACE_DIR", + "EVOSCIENTIST_UI_BACKEND", ]: monkeypatch.delenv(key, raising=False) @@ -76,6 +78,7 @@ class TestEvoScientistConfig: assert config.max_concurrent == 3 assert config.max_iterations == 3 assert config.show_thinking is True + assert config.ui_backend == "rich" assert config.imessage_enabled is False assert config.imessage_allowed_senders == "" @@ -280,8 +283,14 @@ class TestPriorityChain: def test_defaults_when_nothing_set(self, temp_config_dir, clean_env, monkeypatch): """Test defaults are used when nothing is configured.""" # Ensure no env vars affect the test - for key in ["ANTHROPIC_API_KEY", "OPENAI_API_KEY", "TAVILY_API_KEY", - "EVOSCIENTIST_DEFAULT_MODE", "EVOSCIENTIST_WORKSPACE_DIR"]: + for key in [ + "ANTHROPIC_API_KEY", + "OPENAI_API_KEY", + "TAVILY_API_KEY", + "EVOSCIENTIST_DEFAULT_MODE", + "EVOSCIENTIST_WORKSPACE_DIR", + "EVOSCIENTIST_UI_BACKEND", + ]: monkeypatch.delenv(key, raising=False) config = get_effective_config() assert config.provider == "anthropic" @@ -317,6 +326,13 @@ class TestPriorityChain: config = get_effective_config(cli_overrides={"model": "claude-opus-4-5"}) assert config.model == "claude-opus-4-5" + def test_env_ui_backend_override(self, temp_config_dir, monkeypatch): + """UI backend can be selected via environment variable.""" + save_config(EvoScientistConfig(ui_backend="rich")) + monkeypatch.setenv("EVOSCIENTIST_UI_BACKEND", "textual") + config = get_effective_config() + assert config.ui_backend == "textual" + def test_env_api_key_override(self, temp_config_dir, monkeypatch): """Test API keys from env override file.""" save_config(EvoScientistConfig(anthropic_api_key="file-key")) diff --git a/tests/test_imports.py b/tests/test_imports.py deleted file mode 100644 index 85d2f53..0000000 --- a/tests/test_imports.py +++ /dev/null @@ -1,36 +0,0 @@ -"""Smoke tests verifying package structure is intact.""" - - -def test_import_stream_utils(): - from EvoScientist.stream.utils import ( - is_success, - ) - assert callable(is_success) - - -def test_import_stream_emitter(): - from EvoScientist.stream.emitter import StreamEventEmitter - assert callable(StreamEventEmitter.thinking) - - -def test_import_stream_tracker(): - from EvoScientist.stream.tracker import ToolCallTracker - assert ToolCallTracker is not None - - -def test_import_backends(): - from EvoScientist.backends import ( - validate_command, - ) - assert callable(validate_command) - - -def test_import_prompts(): - from EvoScientist.prompts import get_system_prompt - assert callable(get_system_prompt) - - -def test_import_tools(): - from EvoScientist.tools import think_tool - assert think_tool is not None - assert hasattr(think_tool, "invoke") diff --git a/tests/test_message_bus.py b/tests/test_message_bus.py deleted file mode 100644 index 99eefa8..0000000 --- a/tests/test_message_bus.py +++ /dev/null @@ -1,105 +0,0 @@ -"""Tests for the Message Bus decoupling layer.""" - -import asyncio - - -from EvoScientist.channels.bus.events import InboundMessage, OutboundMessage -from EvoScientist.channels.bus.message_bus import MessageBus - - -from tests.conftest import run_async as _run - - -# ── Event tests ── - - -class TestInboundMessage: - def test_session_key(self): - msg = InboundMessage( - channel="telegram", sender_id="u1", - chat_id="c1", content="hi", - ) - assert msg.session_key == "telegram:c1" - - def test_defaults(self): - msg = InboundMessage( - channel="discord", sender_id="u2", - chat_id="c2", content="hello", - ) - assert msg.media == [] - assert msg.metadata == {} - assert msg.message_id == "" - - -class TestOutboundMessage: - def test_fields(self): - msg = OutboundMessage( - channel="telegram", chat_id="c1", content="reply", - ) - assert msg.channel == "telegram" - assert msg.chat_id == "c1" - assert msg.reply_to is None - assert msg.media == [] - - -# ── MessageBus tests ── - - -class TestMessageBus: - def test_inbound_publish_consume(self): - async def _test(): - bus = MessageBus() - msg = InboundMessage( - channel="telegram", sender_id="u1", - chat_id="c1", content="hello", - ) - await bus.publish_inbound(msg) - assert bus.inbound_size == 1 - got = await bus.consume_inbound() - assert got is msg - assert bus.inbound_size == 0 - _run(_test()) - - def test_outbound_publish_consume(self): - async def _test(): - bus = MessageBus() - msg = OutboundMessage( - channel="discord", chat_id="c1", content="reply", - ) - await bus.publish_outbound(msg) - assert bus.outbound_size == 1 - got = await bus.consume_outbound() - assert got is msg - assert bus.outbound_size == 0 - _run(_test()) - - def test_subscribe_and_dispatch(self): - async def _test(): - bus = MessageBus() - received = [] - - async def callback(msg): - received.append(msg) - - bus.subscribe_outbound("telegram", callback) - - msg = OutboundMessage( - channel="telegram", chat_id="c1", content="hi", - ) - await bus.publish_outbound(msg) - - dispatch = asyncio.create_task(bus.dispatch_outbound()) - await asyncio.sleep(0.05) - bus.stop() - await asyncio.sleep(0.05) - dispatch.cancel() - - assert len(received) == 1 - assert received[0] is msg - _run(_test()) - - def test_stop(self): - bus = MessageBus() - assert bus._running is False - bus.stop() - assert bus._running is False diff --git a/tests/test_onboard.py b/tests/test_onboard.py index eae452a..318bc3a 100644 --- a/tests/test_onboard.py +++ b/tests/test_onboard.py @@ -22,10 +22,10 @@ from EvoScientist.config import EvoScientistConfig class TestConstants: - def test_steps_has_nine_items(self): - """Test that STEPS contains exactly 9 steps.""" - assert len(STEPS) == 9 - assert STEPS == ["Provider", "API Key", "Model", "Tavily Key", "Workspace", "Parameters", "Skills", "MCP Servers", "Channels"] + def test_steps_has_ten_items(self): + """Test that STEPS contains exactly 10 steps.""" + assert len(STEPS) == 10 + assert STEPS == ["UI", "Provider", "API Key", "Model", "Tavily Key", "Workspace", "Parameters", "Skills", "MCP Servers", "Channels"] def test_wizard_style_is_style_instance(self): """Test that WIZARD_STYLE is a prompt_toolkit Style.""" @@ -749,6 +749,7 @@ class TestRunOnboard: # Mock all questionary calls mock_q.select.return_value.ask.side_effect = [ + "rich", # UI backend "anthropic", # Provider "claude-sonnet-4-5", # Model "daemon", # Workspace mode @@ -801,6 +802,7 @@ class TestRunOnboard: mock_load.return_value = EvoScientistConfig() mock_q.select.return_value.ask.side_effect = [ + "rich", # UI backend "anthropic", "claude-sonnet-4-5", "daemon", diff --git a/tests/test_tui_widgets.py b/tests/test_tui_widgets.py new file mode 100644 index 0000000..36bf4ef --- /dev/null +++ b/tests/test_tui_widgets.py @@ -0,0 +1,429 @@ +"""Unit tests for TUI widgets. + +Tests widget construction, state transitions, and public APIs +without requiring a running Textual app (no Textual pilot needed). +""" + +from __future__ import annotations + +import importlib +import unittest + + +# --------------------------------------------------------------------------- +# Textual might not be installed — skip entire module if missing +# --------------------------------------------------------------------------- +_has_textual = importlib.util.find_spec("textual") is not None + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestLoadingWidget(unittest.TestCase): + """LoadingWidget construction and attributes.""" + + def test_construction(self): + from EvoScientist.cli.widgets.loading_widget import LoadingWidget + + w = LoadingWidget() + assert w._frame == 0 + assert w._elapsed == 0.0 + assert w._timer_handle is None + + def test_spinner_frames_not_empty(self): + from EvoScientist.cli.widgets.loading_widget import _SPINNER_FRAMES + + assert len(_SPINNER_FRAMES) > 0 + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestThinkingWidget(unittest.TestCase): + """ThinkingWidget construction, append, finalize.""" + + def test_construction_visible(self): + from EvoScientist.cli.widgets.thinking_widget import ThinkingWidget + + w = ThinkingWidget(show_thinking=True) + assert w._is_active is True + assert w._content == "" + assert w.display is True + + def test_construction_hidden(self): + from EvoScientist.cli.widgets.thinking_widget import ThinkingWidget + + w = ThinkingWidget(show_thinking=False) + assert w.display is False + + def test_append_text_accumulates(self): + from EvoScientist.cli.widgets.thinking_widget import ThinkingWidget + + w = ThinkingWidget(show_thinking=True) + w._content = "" # direct access for unit test + # Simulate append (without DOM) + w._content += "hello " + w._content += "world" + assert w._content == "hello world" + + def test_finalize_sets_inactive(self): + from EvoScientist.cli.widgets.thinking_widget import ThinkingWidget + + w = ThinkingWidget(show_thinking=True) + w._is_active = True + # Simulate finalize without DOM + w._is_active = False + assert w._is_active is False + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestAssistantMessage(unittest.TestCase): + """AssistantMessage construction.""" + + def test_empty_construction(self): + from EvoScientist.cli.widgets.assistant_message import AssistantMessage + + w = AssistantMessage() + assert w._content == "" + + def test_initial_content(self): + from EvoScientist.cli.widgets.assistant_message import AssistantMessage + + w = AssistantMessage("hello world") + assert w._content == "hello world" + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestToolCallWidget(unittest.TestCase): + """ToolCallWidget construction and state transitions.""" + + def test_construction(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("read_file", {"path": "/foo.py"}, "abc-123") + assert w._tool_name == "read_file" + assert w._tool_args == {"path": "/foo.py"} + assert w._tool_id == "abc-123" + assert w._status == "running" + + def test_tool_id_property(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("grep", {"pattern": "foo"}, "xyz") + assert w.tool_id == "xyz" + assert w.tool_name == "grep" + + def test_status_transitions(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("execute", {"command": "ls"}) + assert w._status == "running" + # Direct state change for unit test (no DOM) + w._status = "success" + assert w._status == "success" + w._status = "error" + assert w._status == "error" + + def test_result_summary_truncation(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("execute", {}) + w._result_content = "a" * 100 + summary = w._result_summary() + assert len(summary) <= 61 # 57 + "…" + + def test_result_summary_empty(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("execute", {}) + w._result_content = "" + assert w._result_summary() == "done" + + def test_should_collapse_short(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("read_file", {}) + w._result_content = "line1\nline2\nline3" + assert w._should_collapse() is False + + def test_should_collapse_long(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("read_file", {}) + w._result_content = "\n".join(f"line{i}" for i in range(20)) + assert w._should_collapse() is True + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestSubAgentWidget(unittest.TestCase): + """SubAgentWidget construction and name display.""" + + def test_construction(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + w = SubAgentWidget("research-agent", "Search literature") + assert w._sa_name == "research-agent" + assert w._description == "Search literature" + assert w._is_active is True + assert w._tool_count == 0 + + def test_sa_name_property(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + w = SubAgentWidget("code-agent") + assert w.sa_name == "code-agent" + + def test_display_name_with_description(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + w = SubAgentWidget("research-agent", "Search for relevant papers") + name = w._display_name() + assert "Cooking with research-agent" in name + assert "Search for relevant papers" in name + + def test_display_name_truncation(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + long_desc = "A" * 100 + w = SubAgentWidget("agent", long_desc) + name = w._display_name() + assert len(name) < 100 # Should be truncated + + def test_finalize(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + w = SubAgentWidget("agent") + w._is_active = True + w._is_active = False # Simulate finalize + assert w._is_active is False + + def test_update_name(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + w = SubAgentWidget("sub-agent") + assert w._sa_name == "sub-agent" + assert w._description == "" + # Simulate name resolution + w.update_name("planner-agent", "Plan the experiment") + assert w._sa_name == "planner-agent" + assert w._description == "Plan the experiment" + assert "planner-agent" in w._display_name() + assert "Plan the experiment" in w._display_name() + + def test_update_name_preserves_description(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + w = SubAgentWidget("sub-agent", "existing desc") + w.update_name("research-agent") + assert w._sa_name == "research-agent" + # Empty description should not overwrite existing + assert w._description == "existing desc" + + def test_update_name_overwrites_description(self): + from EvoScientist.cli.widgets.subagent_widget import SubAgentWidget + + w = SubAgentWidget("sub-agent", "old desc") + w.update_name("code-agent", "new desc") + assert w._description == "new desc" + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestTodoWidget(unittest.TestCase): + """TodoWidget construction.""" + + def test_construction_empty(self): + from EvoScientist.cli.widgets.todo_widget import TodoWidget + + w = TodoWidget() + assert w._items == [] + + def test_construction_with_items(self): + from EvoScientist.cli.widgets.todo_widget import TodoWidget + + items = [ + {"content": "Search papers", "status": "done"}, + {"content": "Analyze data", "status": "active"}, + {"content": "Write report", "status": "todo"}, + ] + w = TodoWidget(items) + assert len(w._items) == 3 + + def test_update_items(self): + from EvoScientist.cli.widgets.todo_widget import TodoWidget + + w = TodoWidget() + items = [{"content": "task1", "status": "todo"}] + w._items = items # Direct set for unit test + assert w._items == items + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestUserMessage(unittest.TestCase): + """UserMessage construction.""" + + def test_construction(self): + from EvoScientist.cli.widgets.user_message import UserMessage + + w = UserMessage("hello world") + # Should create without error + assert w is not None + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestSystemMessage(unittest.TestCase): + """SystemMessage construction.""" + + def test_construction_default_style(self): + from EvoScientist.cli.widgets.system_message import SystemMessage + + w = SystemMessage("info text") + assert w is not None + + def test_construction_custom_style(self): + from EvoScientist.cli.widgets.system_message import SystemMessage + + w = SystemMessage("error!", msg_style="red") + assert w is not None + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestIsFinalResponse(unittest.TestCase): + """Test _is_final_response helper.""" + + def test_empty_state_is_final(self): + from EvoScientist.cli.tui_interactive import _is_final_response + from EvoScientist.stream.state import StreamState + + state = StreamState() + assert _is_final_response(state) is True + + def test_pending_tools_not_final(self): + from EvoScientist.cli.tui_interactive import _is_final_response + from EvoScientist.stream.state import StreamState + + state = StreamState() + state.tool_calls = [{"name": "read_file", "args": {}}] + state.tool_results = [] # No results yet + assert _is_final_response(state) is False + + def test_all_tools_done_is_final(self): + from EvoScientist.cli.tui_interactive import _is_final_response + from EvoScientist.stream.state import StreamState + + state = StreamState() + state.tool_calls = [{"name": "read_file", "args": {}}] + state.tool_results = [{"name": "read_file", "content": "[OK]"}] + assert _is_final_response(state) is True + + def test_active_subagent_not_final(self): + from EvoScientist.cli.tui_interactive import _is_final_response + from EvoScientist.stream.state import StreamState, SubAgentState + + state = StreamState() + sa = SubAgentState("research-agent") + sa.is_active = True + state.subagents = [sa] + assert _is_final_response(state) is False + + def test_completed_subagent_is_final(self): + from EvoScientist.cli.tui_interactive import _is_final_response + from EvoScientist.stream.state import StreamState, SubAgentState + + state = StreamState() + sa = SubAgentState("research-agent") + sa.is_active = False + state.subagents = [sa] + assert _is_final_response(state) is True + + def test_internal_tools_ignored(self): + from EvoScientist.cli.tui_interactive import _is_final_response + from EvoScientist.stream.state import StreamState + + state = StreamState() + state.tool_calls = [{"name": "ExtractedMemory", "args": {}}] + # No result for internal tool -- should still be considered final + assert _is_final_response(state) is True + + def test_processing_not_final(self): + from EvoScientist.cli.tui_interactive import _is_final_response + from EvoScientist.stream.state import StreamState + + state = StreamState() + state.is_processing = True + assert _is_final_response(state) is False + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestToolCallWidgetIcons(unittest.TestCase): + """ToolCallWidget uses correct status icons (✓/✗/●).""" + + def test_success_icon(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("read_file", {"path": "/f"}, "id1") + # After success, status should be "success" + w._status = "success" + assert w._status == "success" + + def test_error_icon(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("execute", {"command": "bad"}, "id2") + w._status = "error" + assert w._status == "error" + + def test_running_icon(self): + from EvoScientist.cli.widgets.tool_call_widget import ToolCallWidget + + w = ToolCallWidget("grep", {}, "id3") + assert w._status == "running" + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestResponseStripping(unittest.TestCase): + """Response text trailing '...' should be stripped.""" + + def test_strip_trailing_dots(self): + text = "Hello world\n..." + clean = text.strip() + while clean.endswith("\n...") or clean.rstrip() == "...": + clean = clean.rstrip().removesuffix("...").rstrip() + assert clean == "Hello world" + + def test_no_strip_normal_text(self): + text = "Hello world" + clean = text.strip() + while clean.endswith("\n...") or clean.rstrip() == "...": + clean = clean.rstrip().removesuffix("...").rstrip() + assert clean == "Hello world" + + def test_strip_standalone_dots(self): + text = "..." + clean = text.strip() + while clean.endswith("\n...") or clean.rstrip() == "...": + clean = clean.rstrip().removesuffix("...").rstrip() + assert clean == "" + + +@unittest.skipUnless(_has_textual, "textual not installed") +class TestWidgetImports(unittest.TestCase): + """Verify all widgets are importable from the package.""" + + def test_all_imports(self): + from EvoScientist.cli.widgets import ( + LoadingWidget, + ThinkingWidget, + AssistantMessage, + ToolCallWidget, + SubAgentWidget, + TodoWidget, + UserMessage, + SystemMessage, + ) + # All should be classes + for cls in ( + LoadingWidget, ThinkingWidget, AssistantMessage, + ToolCallWidget, SubAgentWidget, TodoWidget, + UserMessage, SystemMessage, + ): + assert isinstance(cls, type), f"{cls} is not a class" + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ui_runtime.py b/tests/test_ui_runtime.py new file mode 100644 index 0000000..dd6b7e1 --- /dev/null +++ b/tests/test_ui_runtime.py @@ -0,0 +1,58 @@ +"""Tests for UI backend runtime selection.""" + +from dataclasses import dataclass + +from EvoScientist.cli.tui_runtime import normalize_ui_backend, resolve_ui_backend, run_streaming + + +def test_normalize_ui_backend_defaults_to_rich(): + assert normalize_ui_backend(None) == "rich" + assert normalize_ui_backend("") == "rich" + + +def test_normalize_ui_backend_accepts_known_values(): + assert normalize_ui_backend("rich") == "rich" + assert normalize_ui_backend("textual") == "textual" + assert normalize_ui_backend("TeXtUaL") == "textual" + + +def test_normalize_ui_backend_unknown_falls_back_to_rich(): + assert normalize_ui_backend("unknown-ui") == "rich" + + +def test_resolve_ui_backend_falls_back_when_textual_unavailable(monkeypatch): + monkeypatch.setattr("EvoScientist.cli.tui_runtime._has_textual_support", lambda: False) + assert resolve_ui_backend("textual") == "rich" + + +def test_resolve_ui_backend_keeps_textual_when_available(monkeypatch): + monkeypatch.setattr("EvoScientist.cli.tui_runtime._has_textual_support", lambda: True) + assert resolve_ui_backend("textual") == "textual" + + +@dataclass +class _BrokenBackend: + name: str = "textual" + + def run_streaming(self, **kwargs): # noqa: ANN003, ANN201 + raise RuntimeError("boom") + + +def test_run_streaming_falls_back_to_rich_on_runtime_error(monkeypatch): + monkeypatch.setattr("EvoScientist.cli.tui_runtime.get_backend", lambda *a, **k: _BrokenBackend()) + + class _RichStub: + def run_streaming(self, **kwargs): # noqa: ANN003, ANN201 + return "fallback-ok" + + monkeypatch.setattr("EvoScientist.cli.tui_runtime.RichStreamingBackend", lambda: _RichStub()) + + result = run_streaming( + ui_backend="textual", + agent=object(), + message="hello", + thread_id="t1", + show_thinking=False, + interactive=True, + ) + assert result == "fallback-ok"