feat: add UserMessage widget and UI backend selection during onboarding
- Introduced UserMessage widget for displaying user input with a styled prompt. - Updated onboarding steps to include UI backend selection (Rich CLI or Textual TUI). - Modified EvoScientistConfig to store selected UI backend. - Enhanced configuration handling to support UI backend environment variable. - Updated README with new UI backend options and commands. - Added tests for new UI backend functionality and UserMessage widget. - Removed obsolete test files and ensured existing tests are updated accordingly.
This commit is contained in:
X-iZhang
@@ -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 .
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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 "",
|
||||
}
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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))
|
||||
@@ -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()
|
||||
@@ -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),
|
||||
))
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -173,9 +173,14 @@ EvoSci # or EvoScientist
|
||||
--use-cwd Use current working directory as workspace
|
||||
--thread-id <id> Resume a conversation thread
|
||||
--no-thinking Disable thinking display
|
||||
--ui <backend> UI backend: rich (default) or textual (beta)
|
||||
-p, --prompt <q> Single-shot mode: execute query and exit
|
||||
```
|
||||
|
||||
> [!NOTE]
|
||||
> In `--ui textual` mode, the built-in TUI commands are:
|
||||
> `/help`, `/current`, `/new`, `/clear`, `/threads`, `/resume <id>`, `/delete <id>`, `/exit`.
|
||||
|
||||
|
||||
|
||||
**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 <source>` | Install a skill from local path or GitHub |
|
||||
|
||||
@@ -35,6 +35,7 @@ dependencies = [
|
||||
"markdownify>=0.14",
|
||||
"nest-asyncio>=1.6",
|
||||
"langchain-mcp-adapters>=0.1",
|
||||
"textual>=0.80",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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"])
|
||||
+18
-2
@@ -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"))
|
||||
|
||||
@@ -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")
|
||||
@@ -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
|
||||
@@ -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",
|
||||
|
||||
@@ -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()
|
||||
@@ -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"
|
||||
Reference in New Issue
Block a user