ouyangbo/EvoScientist-Multi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor/cli command manager (#178)

Browse Source 
* feat(cli): migrate command handling to CommandManager and enhance UI interactions

* feat(cli): add /clear and /help commands to enhance user experience

* feat(cli): implement /new and /resume commands with interactive session management

* Refactor MCP and Skills Command Handling

- Moved the interactive picker style to a centralized widget for consistency across MCP and Skills commands.
- Updated the MCP command to remove the old command dispatch logic, delegating to the new InstallMCPCommand.
- Enhanced the Skills command to utilize a new interactive picker for skill selection, improving user experience.
- Implemented cancellation handling in the picker to differentiate between user cancellations and empty selections.
- Added comprehensive tests for the new command structures and picker functionalities to ensure reliability.

* refactor(cli): streamline CommandManager dispatch and remove deprecated command set

* refactor(cli): enhance error handling and state management in ChannelCommand and RichCLICommandUI

* refactor(cli): update lifecycle callback terminology and improve async prompt handling in RichCLICommandUI

* refactor(cli): enhance SlashCommandCompleter to dynamically fetch workspace directory for autocompletion

* refactor(cli): unify quit handling in RichCLICommandUI with shared _stop helper

* refactor(cli): remove hardcoded slash commands and utilize command manager for dynamic completion

* refactor(cli): update MCP and skills command files for improved clarity and organization

* refactor(mcp_ui): remove unnecessary newline in _show_mcp_config function
This commit is contained in:
Xi Zhang
2026-04-24 11:38:44 +02:00
committed by GitHub
parent f704e3d761
commit 92df3e1844
24 changed files with 1779 additions and 957 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EvoScientist/cli/channel.py
-120
View File
@@ -19,7 +19,6 @@ from dataclasses import dataclass
from typing import Any
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from ..stream.console import console
@@ -682,125 +681,6 @@ def _print_channel_panel(channels: list[tuple[str, bool, str]]) -> None:
console.print()
def _cmd_channel(
args: str,
agent: Any,
thread_id: str,
*,
send_thinking: bool | None = None,
) -> None:
"""Start a channel in background using bus mode.
Usage:
/channel [telegram|discord|imessage] -- start channel (default from config)
/channel status -- show current channel status
/channel stop -- stop running channel
"""
global _cli_agent, _cli_thread_id
from ..config import load_config
app_config = load_config()
channel_type = args.strip().lower() if args and args.strip() else ""
if channel_type == "status":
running = _channels_running_list()
if running and _manager:
detailed = _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)
console.print(table)
console.print()
else:
console.print("[dim]No channel running[/dim]\n")
return
if not channel_type:
channel_type = app_config.channel_enabled
if not channel_type:
console.print("[yellow]No channel configured.[/yellow]")
console.print(
"[dim]Run[/dim] evosci onboard [dim]or specify:[/dim] /channel telegram\n"
)
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=send_thinking,
)
results.append((ct, True, "connected (bus)"))
except Exception as e:
results.append((ct, False, str(e)))
_print_channel_panel(results)
return
_cli_agent = agent
_cli_thread_id = thread_id
# Override channel_enabled for this invocation
original = app_config.channel_enabled
app_config.channel_enabled = channel_type
try:
_start_channels_bus_mode(
app_config,
agent,
thread_id,
send_thinking=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
_print_channel_panel(results)
def _cmd_channel_stop(channel_type: str | None = None) -> None:
"""Stop background channel(s).
Args:
channel_type: Specific channel to stop, or None to stop all.
"""
if not _channels_is_running():
console.print("[dim]No channel running[/dim]\n")
return
if channel_type:
if not _channels_is_running(channel_type):
console.print(f"[dim]{channel_type} is not running[/dim]\n")
return
_channels_stop(channel_type)
console.print(f"[dim]{channel_type} stopped[/dim]\n")
else:
running = _channels_running_list()
_channels_stop()
console.print(f"[dim]{', '.join(running)} stopped[/dim]\n")
def _auto_start_channel(
agent: Any,
thread_id: str,
EvoScientist/cli/interactive.py
+161 -339
View File
@@ -5,6 +5,7 @@ import logging
import queue
import random
import sys
from collections.abc import Callable
from datetime import datetime
from typing import Any
@@ -26,19 +27,17 @@ from prompt_toolkit.styles import Style as PtStyle # type: ignore[import-untype
from rich.markdown import Markdown
from rich.markup import escape
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
import EvoScientist.cli.channel as _ch_mod
from ..commands.base import CommandContext
from ..commands.manager import manager as cmd_manager
from ..sessions import (
_format_relative_time,
delete_thread,
generate_thread_id,
get_checkpointer,
get_thread_messages,
get_thread_metadata,
list_threads,
resolve_thread_id_prefix,
thread_exists,
)
@@ -50,19 +49,11 @@ from .channel import (
ChannelMessage,
_auto_start_channel,
_channels_is_running,
_cmd_channel,
_cmd_channel_stop,
_message_queue,
_set_channel_response,
)
from .file_mentions import complete_file_mention, resolve_file_mentions
from .mcp_ui import _cmd_mcp
from .skills_cmd import (
_cmd_install_skill,
_cmd_install_skills,
_cmd_list_skills,
_cmd_uninstall_skill,
)
from .rich_command_ui import RichCLICommandUI
from .status_bar import (
SPINNER_FRAMES,
STATUS_BAD,
@@ -89,7 +80,6 @@ _channel_logger = logging.getLogger(__name__)
# Keeps references to fire-and-forget coroutines so they aren't GC'd mid-flight.
_background_tasks: set[asyncio.Task] = set()
# =============================================================================
# Banner
# =============================================================================
@@ -155,23 +145,6 @@ def print_banner(
# Slash-command completer
# =============================================================================
_SLASH_COMMANDS = [
("/current", "Show current session info"),
("/threads", "List recent sessions"),
("/resume", "Resume a previous session (prefix match)"),
("/delete", "Delete a saved session"),
("/new", "Start a new session"),
("/skills", "List installed skills"),
("/install-skill", "Add a skill from path or GitHub"),
("/uninstall-skill", "Remove an installed skill"),
("/evoskills", "Browse and install EvoSkills (optional: /evoskills <tag>)"),
("/mcp", "Manage MCP servers"),
("/channel", "Configure messaging channels"),
("/compact", "Compact conversation to free context"),
("/model", "Switch model (--save to persist)"),
("/exit", "Quit EvoScientist"),
]
_COMPLETION_STYLE = PtStyle.from_dict(
{
"completion-menu": "bg:default noreverse nounderline noitalic",
@@ -191,32 +164,28 @@ _COMPLETION_STYLE = PtStyle.from_dict(
}
)
# Style for questionary pickers — matches _COMPLETION_STYLE visual language:
# gray (#888888) for non-selected, bold for selected, no background changes.
_PICKER_STYLE = PtStyle.from_dict(
{
"questionmark": "#888888",
"question": "",
"pointer": "bold",
"highlighted": "bold",
"text": "#888888",
"answer": "bold",
}
)
class SlashCommandCompleter(Completer):
"""Autocomplete for slash commands and ``@file`` mentions."""
"""Autocomplete for slash commands and ``@file`` mentions.
def __init__(self, workspace_dir: str | None = None) -> None:
self._workspace_dir = workspace_dir
``workspace_getter`` is invoked on every keystroke so ``@file``
suggestions automatically follow ``/new`` / ``/resume`` workspace
changes without having to poke the completer from the callbacks.
"""
def __init__(
self,
workspace_getter: Callable[[], str | None] | None = None,
) -> None:
self._workspace_getter = workspace_getter or (lambda: None)
def get_completions(self, document, complete_event):
text = document.text_before_cursor
workspace_dir = self._workspace_getter()
# @file mention completion
if "@" in text:
candidates = complete_file_mention(text, self._workspace_dir)
candidates = complete_file_mention(text, workspace_dir)
if candidates:
# Replace from the last '@' token
import re as _re
@@ -230,7 +199,9 @@ class SlashCommandCompleter(Completer):
# Slash command completion
if not text.startswith("/"):
return
for cmd, desc in _SLASH_COMMANDS:
# ``list_commands`` is dedup'd on the Command instance so aliases
# (e.g. /quit, /q for /exit) don't appear as separate rows.
for cmd, desc in sorted(cmd_manager.list_commands()):
if cmd.startswith(text):
yield Completion(
cmd,
@@ -320,7 +291,9 @@ def cmd_interactive(
session = PromptSession(
history=FileHistory(history_file),
auto_suggest=AutoSuggestFromHistory(),
completer=SlashCommandCompleter(workspace_dir=workspace_dir),
completer=SlashCommandCompleter(
workspace_getter=lambda: state["workspace_dir"],
),
complete_style=CompleteStyle.COLUMN,
complete_while_typing=True,
style=_COMPLETION_STYLE,
@@ -372,6 +345,21 @@ def cmd_interactive(
on_progress=_on_mcp_progress,
)
def _on_status_after_compact(input_tokens: int) -> None:
"""Mirror inline /compact post-update: refresh both fields so the
next status render reflects the reduced context immediately.
``_refresh_status_snapshot`` is invoked by the dispatch block once
the command finishes (since it's async)."""
state["status_last_input_tokens"] = input_tokens
state["status_base_snapshot"] = make_usage_status_snapshot(
input_tokens,
model_name=model,
)
# ``rich_ui`` is constructed inside ``_async_main_loop`` so the
# lifecycle callbacks can close over ``checkpointer`` from
# ``get_checkpointer()``.
def _start_agent_load(checkpointer) -> None:
progress_tracker.prime()
agent_loader.start(
@@ -523,39 +511,6 @@ def cmd_interactive(
console.print(f"[red]Thread '{escape(tid)}' not found.[/red]")
return None
async def _cmd_threads():
"""Handle /threads command — show recent sessions."""
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
console.print("[yellow]No saved sessions.[/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 t in threads:
tid = t["thread_id"]
marker = " *" if tid == state["thread_id"] else ""
table.add_row(
f"{tid}{marker}",
t.get("preview", "") or "",
str(t.get("message_count", 0)),
t.get("model", "") or "",
_format_relative_time(t.get("updated_at")),
)
console.print()
console.print(table)
console.print(
"[dim] /resume[/dim] to continue a session [dim]/delete <id>[/dim] to remove [dim]/new[/dim] to start fresh"
)
console.print()
async def _render_history(thread_id: str):
"""Display conversation history for a resumed session."""
messages = await get_thread_messages(thread_id)
@@ -630,120 +585,80 @@ def cmd_interactive(
console.print("[dim]── End of history ──[/dim]")
console.print()
async def _cmd_resume(arg: str, checkpointer):
"""Handle /resume [id] — resume a previous session."""
if not arg:
# Show interactive session picker with conversation previews
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
console.print("[yellow]No sessions to resume.[/yellow]")
return
import questionary
from .widgets.thread_selector import _build_items
choices = []
items = _build_items(threads)
for item in items:
if item["type"] == "header":
choices.append(
questionary.Separator(
f"\u2500\u2500 \U0001f4c2 {item['label']}"
)
)
elif item["type"] == "subheader":
choices.append(questionary.Separator(f" {item['label']}"))
else:
t = item["thread"]
tid = t["thread_id"]
preview = t.get("preview", "") or ""
msgs = t.get("message_count", 0)
model = t.get("model", "") or ""
when = _format_relative_time(t.get("updated_at"))
indent = " " if item.get("indented") else " "
parts = [f"{indent}{tid}"]
if preview:
parts.append(
preview[:40] + "\u2026" if len(preview) > 40 else preview
)
parts.append(f"({msgs} msgs)")
if model:
parts.append(model)
if when:
parts.append(when)
label = " ".join(parts)
choices.append(questionary.Choice(title=label, value=tid))
from prompt_toolkit.layout.dimension import Dimension
from questionary.prompts.common import InquirerControl
prompt = questionary.select(
"Select session to resume:",
choices=choices,
style=_PICKER_STYLE,
)
# Limit visible list to 10 rows with scrolling
for window in prompt.application.layout.find_all_windows():
if isinstance(window.content, InquirerControl):
window.height = Dimension(max=10)
break
selected = prompt.ask()
if selected is None:
return
arg = selected
resolved = await _resolve_thread_id(arg)
if not resolved:
return
meta = await get_thread_metadata(resolved)
ws = (meta or {}).get("workspace_dir", "") or state["workspace_dir"]
state["thread_id"] = resolved
state["resumed"] = True
if ws:
state["workspace_dir"] = ws
state["status_started_at"] = datetime.now()
state["status_last_input_tokens"] = None
# Rebuild the agent in the background so the resumed transcript
# and prompt render immediately; the next message awaits the load.
_start_agent_load(checkpointer)
await _refresh_status_snapshot(reset_streaming_text=True)
console.print(f"[green]Resumed session:[/green] [yellow]{resolved}[/yellow]")
if state["workspace_dir"]:
console.print(
f"[dim]Workspace:[/dim] [cyan]{_shorten_path(state['workspace_dir'])}[/cyan]"
)
console.print()
await _render_history(resolved)
async def _cmd_delete(arg: str):
"""Handle /delete <id> — delete a saved session."""
if not arg:
console.print("[red]Usage: /delete <thread-id>[/red]")
return
resolved = await _resolve_thread_id(arg)
if not resolved:
return
if resolved == state["thread_id"]:
console.print("[red]Cannot delete the current session.[/red]")
return
deleted = await delete_thread(resolved)
if deleted:
console.print(f"[green]Deleted session {resolved}.[/green]")
else:
console.print(f"[red]Session {resolved} not found.[/red]")
async def _async_main_loop():
"""Async main loop with prompt_async and channel queue checking."""
nonlocal model
async with get_checkpointer() as checkpointer:
# Lifecycle callbacks (new / resume) need ``checkpointer``
# in scope — define the ``rich_ui`` adapter here rather than
# at the outer function level.
def _on_start_new_session() -> None:
"""NewCommand callback — rotate workspace (if not fixed),
issue a new thread id, reset session-scoped status fields,
and kick off background agent reload. The dispatch block
refreshes the status bar post-execute (symmetric with
/compact)."""
if not workspace_fixed:
state["workspace_dir"] = _create_session_workspace(run_name)
state["thread_id"] = generate_thread_id()
state["resumed"] = False
state["status_started_at"] = datetime.now()
state["status_last_input_tokens"] = None
_start_agent_load(checkpointer)
console.print(
f"[green]New session:[/green] [yellow]{state['thread_id']}[/yellow]"
)
if state["workspace_dir"]:
console.print(
f"[dim]Workspace:[/dim] [cyan]"
f"{_shorten_path(state['workspace_dir'])}[/cyan]\n"
)
async def _on_handle_session_resume(
thread_id: str, workspace_dir: str | None
) -> None:
"""ResumeCommand callback — after the command resolves
the thread id + restores workspace from metadata, this
callback mutates REPL state, reloads the agent, and
renders conversation history."""
if workspace_dir:
state["workspace_dir"] = workspace_dir
state["thread_id"] = thread_id
state["resumed"] = True
state["status_started_at"] = datetime.now()
state["status_last_input_tokens"] = None
_start_agent_load(checkpointer)
await _refresh_status_snapshot(reset_streaming_text=True)
console.print(
f"[green]Resumed session:[/green] [yellow]{thread_id}[/yellow]"
)
if state["workspace_dir"]:
console.print(
f"[dim]Workspace:[/dim] [cyan]"
f"{_shorten_path(state['workspace_dir'])}[/cyan]"
)
console.print()
await _render_history(thread_id)
# Rich CLI collapses ``request_quit`` / ``force_quit`` into the
# same "break the prompt loop" effect — there's no equivalent
# of the TUI's double-press-to-confirm quit distinction. A
# shared ``_stop`` helper makes the intentional symmetry
# explicit instead of silently duplicating a lambda.
def _stop() -> None:
state["running"] = False
rich_ui = RichCLICommandUI(
console,
on_request_quit=_stop,
on_force_quit=_stop,
on_clear_chat=lambda: console.clear(),
on_status_after_compact=_on_status_after_compact,
on_start_new_session=_on_start_new_session,
on_handle_session_resume=_on_handle_session_resume,
)
# Handle --thread-id resume
if thread_id:
resolved = await _resolve_thread_id(thread_id)
@@ -1036,168 +951,75 @@ def cmd_interactive(
_print_separator()
# Special commands
if user_input.lower() in ("/exit", "/quit", "/q"):
state["running"] = False
break
if user_input.lower() == "/threads":
await _cmd_threads()
continue
if user_input.lower().startswith("/resume"):
arg = user_input[len("/resume") :].strip()
await _cmd_resume(arg, checkpointer)
continue
if user_input.lower().startswith("/delete"):
arg = user_input[len("/delete") :].strip()
await _cmd_delete(arg)
continue
if user_input.lower() == "/new":
# New session: new thread; workspace only changes if not fixed
if not workspace_fixed:
state["workspace_dir"] = _create_session_workspace(
run_name
)
state["thread_id"] = generate_thread_id()
state["resumed"] = False
state["status_started_at"] = datetime.now()
state["status_last_input_tokens"] = None
# Background agent reload — next message awaits it.
_start_agent_load(checkpointer)
await _refresh_status_snapshot(reset_streaming_text=True)
console.print(
f"[green]New session:[/green] [yellow]{state['thread_id']}[/yellow]"
)
if state["workspace_dir"]:
console.print(
f"[dim]Workspace:[/dim] [cyan]{_shorten_path(state['workspace_dir'])}[/cyan]\n"
)
continue
if user_input.lower() == "/current":
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]"
)
if memory_dir:
console.print(
f"[dim]Memory dir:[/dim] [cyan]{_shorten_path(memory_dir)}[/cyan]"
)
console.print()
continue
if user_input.lower() == "/skills":
_cmd_list_skills()
continue
if user_input.lower().startswith("/install-skill"):
source = user_input[len("/install-skill") :].strip()
_cmd_install_skill(source)
continue
if user_input.lower().startswith("/uninstall-skill"):
name = user_input[len("/uninstall-skill") :].strip()
_cmd_uninstall_skill(name)
continue
if user_input.lower().startswith("/evoskills"):
browse_args = user_input[len("/evoskills") :].strip()
_cmd_install_skills(browse_args)
continue
if user_input.lower().startswith("/mcp"):
_cmd_mcp(user_input[len("/mcp") :])
continue
if user_input.lower().startswith("/channel"):
args = user_input[len("/channel") :].strip()
if args.lower().startswith("stop"):
stop_arg = args[len("stop") :].strip()
_cmd_channel_stop(stop_arg or None)
else:
await _await_agent_ready()
_cmd_channel(
args,
agent_loader.agent,
state["thread_id"],
send_thinking=channel_send_thinking,
)
continue
if user_input.lower() == "/compact":
from .commands import (
build_compact_summary_renderable,
compact_conversation,
render_compact_result,
)
await _await_agent_ready()
with console.status(
"[cyan]Compacting conversation...[/cyan]"
):
result = await compact_conversation(
agent=agent_loader.agent,
thread_id=state["thread_id"],
input_tokens_hint=state.get(
"status_last_input_tokens"
),
)
console.print(render_compact_result(result))
summary_renderable = build_compact_summary_renderable(
result
)
if summary_renderable is not None:
console.print(summary_renderable)
if result.status == "ok" and result.tokens_after > 0:
state["status_last_input_tokens"] = result.tokens_after
state["status_base_snapshot"] = (
make_usage_status_snapshot(
result.tokens_after,
model_name=model,
)
)
await _refresh_status_snapshot(
reset_streaming_text=True,
)
continue
if user_input.lower().startswith("/model"):
from ..commands.base import CommandContext
from ..commands.manager import manager as cmd_manager
from ..EvoScientist import _ensure_config
from .rich_command_ui import RichCLICommandUI
# /model is ``needs_agent=False`` — it builds its
# own agent — so we don't wait for the current
# load; /model is the way to fix a broken one.
# ==== Shared CommandManager dispatch ====
# Every registered slash command routes through the
# manager. Unresolved input (non-slash, typo) falls
# through to the agent message path below.
_parsed = cmd_manager.resolve(user_input)
if _parsed is not None:
_cmd, _cmd_args = _parsed
_agent_for_ctx: Any = agent_loader.agent
if _cmd.needs_agent(_cmd_args):
_agent_for_ctx = await _await_agent_ready()
ctx = CommandContext(
agent=None,
agent=_agent_for_ctx,
thread_id=state["thread_id"],
ui=RichCLICommandUI(console),
ui=rich_ui,
workspace_dir=state["workspace_dir"],
checkpointer=checkpointer,
config=config,
input_tokens_hint=state.get("status_last_input_tokens"),
)
await cmd_manager.execute(user_input, ctx)
if ctx.agent is not None:
# ExitCommand signals quit via ``force_quit`` →
# callback flips ``state["running"]`` to False.
if not state["running"]:
break
# Agent swap (e.g. /model successfully built a
# new agent): adopt into loader + reset status
# snapshot + sync channel globals.
agent_swapped = (
ctx.agent is not None
and ctx.agent is not _agent_for_ctx
)
if agent_swapped:
from ..EvoScientist import _ensure_config
agent_loader.adopt(ctx.agent)
cfg = _ensure_config()
model = cfg.model
state["status_base_snapshot"] = (
make_empty_status_snapshot(model)
)
await _refresh_status_snapshot(
reset_streaming_text=True,
)
if _channels_is_running():
_ch_mod._cli_agent = ctx.agent
_ch_mod._cli_thread_id = state["thread_id"]
# Commands that mutate status fields need an
# async refresh here (/compact + /new use sync
# callbacks; /model swaps the agent). /resume
# awaits its own refresh inline inside the
# async callback.
if agent_swapped or _cmd.name in ("/compact", "/new"):
await _refresh_status_snapshot(
reset_streaming_text=True,
)
continue
# Unknown slash command (typo) — short-circuit so
# it doesn't get forwarded to the agent, which
# would waste tokens interpreting the nonsense.
if user_input.lstrip().startswith("/"):
bad_cmd = user_input.split(None, 1)[0]
console.print(
f"[red]Unknown command:[/red] {escape(bad_cmd)}"
)
console.print(
"[dim]Type /help to see available commands.[/dim]"
)
console.print()
continue
# Resolve @file mentions — inject file contents inline
EvoScientist/cli/mcp_install_cmd.py
+3 -3
View File
@@ -21,7 +21,7 @@ from ..mcp.registry import (
install_mcp_servers,
)
from ..stream.console import console
from .interactive import _PICKER_STYLE
from .widgets.thread_selector import PICKER_STYLE
_INSTALLED_INDICATOR = ("fg:#4caf50", "\u2713 ")
@@ -46,7 +46,7 @@ def _checkbox_ask(choices, message: str, **kwargs):
return questionary.checkbox(
message,
choices=choices,
style=_PICKER_STYLE,
style=PICKER_STYLE,
qmark="\u276f",
**kwargs,
).ask()
@@ -89,7 +89,7 @@ def _browse_and_select(
selected_tag = questionary.select(
"Filter by tag:",
choices=tag_choices,
style=_PICKER_STYLE,
style=PICKER_STYLE,
qmark="\u276f",
).ask()
EvoScientist/cli/mcp_ui.py
+1 -127
View File
@@ -1,4 +1,4 @@
"""MCP server display, operations, and /mcp slash-command dispatcher."""
"""Shared UI helpers for MCP server display and operations (used by the Typer `mcp` commands)."""
from typing import Any
@@ -181,129 +181,3 @@ def _show_mcp_config(name: str = "", *, show_blank_line: bool = True) -> str:
if show_blank_line:
console.print()
return "ok"
def _cmd_mcp_add(args_str: str) -> None:
"""Handle ``/mcp add ...``."""
import shlex
from ..mcp import parse_mcp_add_args
if not args_str.strip():
console.print("[bold]Usage:[/bold] /mcp add <name> <command-or-url> [args...]")
console.print()
console.print(
"[dim]Transport is auto-detected: URLs \u2192 http, commands \u2192 stdio[/dim]"
)
console.print()
console.print("[bold]Examples:[/bold]")
console.print(
" /mcp add sequential-thinking npx -y @modelcontextprotocol/server-sequential-thinking"
)
console.print(" /mcp add docs-langchain https://docs.langchain.com/mcp")
console.print(
" /mcp add my-sse http://localhost:9090/sse --transport sse --expose-to research-agent"
)
console.print()
console.print("[dim]Options:[/dim]")
console.print(" --transport T Transport type (default: auto-detect)")
console.print(
" --tools t1,t2 Tool allowlist (supports wildcards: *_exa, read_*)"
)
console.print(" --expose-to a1,a2 Target agents (default: main)")
console.print(" --header Key:Value HTTP header (repeatable)")
console.print(" --env KEY=VALUE Env var for stdio (repeatable)")
console.print(
" --env-ref KEY Env var as runtime ${KEY} reference (repeatable)"
)
console.print()
return
try:
tokens = shlex.split(args_str)
kwargs = parse_mcp_add_args(tokens)
_mcp_add_server_from_kwargs(kwargs, show_reload_hint=True)
except ValueError as exc:
console.print(f"[red]{exc}[/red]")
console.print()
def _cmd_mcp_edit(args_str: str) -> None:
"""Handle ``/mcp edit <name> --field value ...``."""
import shlex
from ..mcp import parse_mcp_edit_args
if not args_str.strip():
console.print("[bold]Usage:[/bold] /mcp edit <name> --<field> <value> ...")
console.print()
console.print(
"[dim]Fields:[/dim] --transport, --command, --url, --args, --tools, --expose-to, --header, --env"
)
console.print(
"[dim]Use[/dim] --tools none [dim]or[/dim] --expose-to none [dim]to clear a field.[/dim]"
)
console.print()
console.print("[bold]Examples:[/bold]")
console.print(" /mcp edit filesystem --expose-to main,code-agent")
console.print(" /mcp edit filesystem --tools read_file,write_file")
console.print(" /mcp edit my-api --url http://new-host:8080/mcp")
console.print(" /mcp edit my-api --tools none")
console.print()
return
try:
tokens = shlex.split(args_str)
name, fields = parse_mcp_edit_args(tokens)
_mcp_edit_server_fields(name, fields, show_reload_hint=True)
except ValueError as exc:
console.print(f"[red]{exc}[/red]")
console.print()
def _cmd_mcp_remove(name: str) -> None:
"""Handle ``/mcp remove <name>``."""
_mcp_remove_server(name, show_reload_hint=True)
console.print()
def _cmd_mcp_config(name: str) -> None:
"""Handle ``/mcp config [name]``."""
_show_mcp_config(name, show_blank_line=True)
def _cmd_mcp(args: str) -> None:
"""Dispatch ``/mcp`` subcommands."""
args = args.strip()
if not args:
_mcp_list_servers()
return
parts = args.split(maxsplit=1)
subcmd = parts[0].lower()
subargs = parts[1] if len(parts) > 1 else ""
if subcmd == "list":
_mcp_list_servers()
elif subcmd == "add":
_cmd_mcp_add(subargs)
elif subcmd == "edit":
_cmd_mcp_edit(subargs)
elif subcmd == "remove":
_cmd_mcp_remove(subargs)
elif subcmd == "config":
_cmd_mcp_config(subargs)
elif subcmd == "install":
from .mcp_install_cmd import _cmd_install_mcp
_cmd_install_mcp(subargs)
else:
console.print("[bold]MCP commands:[/bold]")
console.print(" /mcp List configured servers")
console.print(" /mcp list List configured servers")
console.print(" /mcp config Show detailed server config")
console.print(" /mcp add ... Add a server")
console.print(" /mcp edit ... Edit an existing server")
console.print(" /mcp remove ... Remove a server")
console.print(" /mcp install ... Browse and install servers")
console.print()
EvoScientist/cli/rich_command_ui.py
+171 -41
View File
@@ -1,12 +1,21 @@
"""CommandUI Protocol adapter for the Rich CLI surface.
Methods not exercised by the currently-migrated commands raise
``NotImplementedError`` rather than silently returning ``None``, so
future callers fail loudly instead of pretending the command ran.
Lifecycle methods (``request_quit``, ``force_quit``, ``clear_chat``,
``start_new_session``, ``handle_session_resume``, ``update_status_after_compact``)
are callback-driven: when their corresponding ``on_*`` constructor kwarg
is ``None``, the method is a silent no-op, mirroring
``ChannelCommandUI``'s fallback pattern. Callers that need a specific
side-effect (REPL quit flag flip, status-bar refresh, …) wire the
callback at construction time; non-interactive surfaces (tests,
alternate REPLs) can leave callbacks unset without crashing.
``wait_for_*`` methods return ``None`` on cancel / fallback and are
always safe to ``await``.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any
from rich.console import Console
@@ -16,17 +25,42 @@ from ..commands.base import CommandUI
class RichCLICommandUI(CommandUI):
"""CommandUI implementation that prints to a Rich ``Console``."""
"""CommandUI implementation that prints to a Rich ``Console``.
def __init__(self, console: Console) -> None:
Commands that affect CLI-closure state (session lifecycle, exit flag,
status-bar snapshot) go through optional callbacks wired by the REPL.
This mirrors ``ChannelCommandUI``'s injection pattern and keeps
``interactive.py``'s ``state`` dict as the single source of truth.
"""
def __init__(
self,
console: Console,
*,
on_request_quit: Callable[[], None] | None = None,
on_force_quit: Callable[[], None] | None = None,
on_clear_chat: Callable[[], None] | None = None,
on_status_after_compact: Callable[[int], None] | None = None,
on_start_new_session: Callable[[], None] | None = None,
on_handle_session_resume: (
Callable[[str, str | None], Awaitable[None]] | None
) = None,
) -> None:
self.console = console
self._on_request_quit = on_request_quit
self._on_force_quit = on_force_quit
self._on_clear_chat = on_clear_chat
self._on_status_after_compact = on_status_after_compact
self._on_start_new_session = on_start_new_session
self._on_handle_session_resume = on_handle_session_resume
# Bound ``console.status(...)`` context manager used by
# /compact's start/stop indicator pair.
self._compact_status_ctx: Any = None
# ── Core I/O ─────────────────────────────────────────────
@property
def supports_interactive(self) -> bool:
# Rich CLI has no picker widget, but wait_for_* fall back to
# printing a table and returning None (see wait_for_model_pick).
return True
def append_system(self, text: str, style: str = "dim") -> None:
@@ -73,54 +107,150 @@ class RichCLICommandUI(CommandUI):
``ctx.agent`` change post-``cmd_manager.execute``."""
return
# ── Not yet migrated ────────────────────────────────────
# ── Interactive pickers ────────────────────────────────
async def wait_for_thread_pick(
self, threads: list[dict], current_thread: str, title: str
) -> str | None:
raise NotImplementedError(
"RichCLICommandUI.wait_for_thread_pick — implement when "
"migrating /threads / /resume"
"""Interactive workspace-grouped thread picker using ``questionary``.
Ported from the pre-migration ``_cmd_resume`` implementation.
Returns the selected ``thread_id`` string, or ``None`` on cancel.
Callers (``ResumeCommand``/``DeleteCommand``) pre-check for
empty thread lists before invoking this method.
"""
import questionary # type: ignore[import-untyped]
from prompt_toolkit.layout.dimension import ( # type: ignore[import-untyped]
Dimension,
)
from questionary.prompts.common import ( # type: ignore[import-untyped]
InquirerControl,
)
from ..sessions import _format_relative_time
from .widgets.thread_selector import PICKER_STYLE, _build_items
choices: list[Any] = []
for item in _build_items(threads):
if item["type"] == "header":
choices.append(questionary.Separator(f"── \U0001f4c2 {item['label']}"))
elif item["type"] == "subheader":
choices.append(questionary.Separator(f" {item['label']}"))
else:
t = item["thread"]
tid = t["thread_id"]
preview = t.get("preview", "") or ""
msgs = t.get("message_count", 0)
model = t.get("model", "") or ""
when = _format_relative_time(t.get("updated_at"))
indent = " " if item.get("indented") else " "
marker = " *" if tid == current_thread else ""
parts = [f"{indent}{tid}{marker}"]
if preview:
parts.append(preview[:40] + "…" if len(preview) > 40 else preview)
parts.append(f"({msgs} msgs)")
if model:
parts.append(model)
if when:
parts.append(when)
label = " ".join(parts)
choices.append(questionary.Choice(title=label, value=tid))
prompt = questionary.select(title, choices=choices, style=PICKER_STYLE)
# Limit visible list to 10 rows with scrolling. Touches
# questionary/prompt-toolkit private internals so guard against
# library-shape changes — picker stays functional at default
# height even if the cap fails.
try:
for window in prompt.application.layout.find_all_windows():
if isinstance(window.content, InquirerControl):
window.height = Dimension(max=10)
break
except Exception:
pass
# ``ask_async`` (questionary >= 2.0.1) avoids blocking the
# asyncio event loop while the user interacts with the picker.
return await prompt.ask_async()
# ── Lifecycle callbacks ───────────────────────────────
def clear_chat(self) -> None:
if self._on_clear_chat is not None:
self._on_clear_chat()
else:
self.console.clear()
def request_quit(self) -> None:
if self._on_request_quit is not None:
self._on_request_quit()
def force_quit(self) -> None:
if self._on_force_quit is not None:
self._on_force_quit()
def start_new_session(self) -> None:
if self._on_start_new_session is not None:
self._on_start_new_session()
async def handle_session_resume(
self, thread_id: str, workspace_dir: str | None = None
) -> None:
if self._on_handle_session_resume is not None:
await self._on_handle_session_resume(thread_id, workspace_dir)
# /compact indicator pair — duck-typed by ``CompactCommand`` via
# ``getattr``, not declared on the ``CommandUI`` Protocol.
def start_compacting_indicator(self) -> None:
# Idempotent: close any lingering context before starting a new
# one so a double-call (e.g. two overlapping /compact attempts
# via the message queue) can't leak a Rich Live handle.
if self._compact_status_ctx is not None:
try:
self._compact_status_ctx.__exit__(None, None, None)
except Exception:
pass
self._compact_status_ctx = None
status = self.console.status("[cyan]Compacting conversation...[/cyan]")
status.__enter__()
self._compact_status_ctx = status
def stop_compacting_indicator(self) -> None:
ctx = self._compact_status_ctx
self._compact_status_ctx = None
if ctx is not None:
try:
ctx.__exit__(None, None, None)
except Exception:
pass
def update_status_after_compact(self, input_tokens: int) -> None:
if self._on_status_after_compact is not None:
self._on_status_after_compact(input_tokens)
# ── Skill / MCP browse (delegated to worker threads) ──
async def wait_for_skill_browse(
self, index: list[dict], installed_names: set[str], pre_filter_tag: str
) -> list[str] | None:
raise NotImplementedError(
"RichCLICommandUI.wait_for_skill_browse — implement when "
"migrating /skills / /evoskills"
"""Delegate to the extracted questionary picker on a worker
thread — questionary blocks the event loop so the call must
not happen on the main asyncio thread."""
import asyncio
from .skills_cmd import _pick_skills_interactive
return await asyncio.to_thread(
_pick_skills_interactive, index, installed_names, pre_filter_tag
)
async def wait_for_mcp_browse(
self, servers: list, installed_names: set[str], pre_filter_tag: str
) -> list | None:
raise NotImplementedError(
"RichCLICommandUI.wait_for_mcp_browse — implement when migrating /mcp"
)
"""Delegate to the MCP browse picker on a worker thread."""
import asyncio
def clear_chat(self) -> None:
raise NotImplementedError(
"RichCLICommandUI.clear_chat — implement when migrating /clear"
)
from .mcp_install_cmd import _browse_and_select
def request_quit(self) -> None:
raise NotImplementedError(
"RichCLICommandUI.request_quit — implement when migrating /exit"
)
def force_quit(self) -> None:
raise NotImplementedError(
"RichCLICommandUI.force_quit — implement when migrating /exit"
)
def start_new_session(self) -> None:
raise NotImplementedError(
"RichCLICommandUI.start_new_session — implement when migrating /new"
)
async def handle_session_resume(
self, thread_id: str, workspace_dir: str | None = None
) -> None:
raise NotImplementedError(
"RichCLICommandUI.handle_session_resume — implement when migrating /resume"
return await asyncio.to_thread(
_browse_and_select, servers, installed_names, pre_filter_tag
)
EvoScientist/cli/skills_cmd.py
+29 -212
View File
@@ -1,160 +1,30 @@
"""Slash commands for skill management: /skills, /install-skill, /uninstall-skill, /evoskills."""
from pathlib import Path
"""Shared UI helpers for skill-management commands (picker used by /evoskills)."""
from ..stream.console import console
from .agent import _shorten_path
def _cmd_list_skills() -> None:
"""List all available skills (workspace, global, and built-in)."""
from ..paths import GLOBAL_SKILLS_DIR, USER_SKILLS_DIR
from ..tools.skills_manager import list_skills
def _pick_skills_interactive(
index: list[dict],
installed_names: set[str],
pre_filter_tag: str,
) -> list[str] | None:
"""Interactive questionary picker for EvoSkills browse.
skills = list_skills(include_system=True)
Two-phase picker:
1. tag filter — ``questionary.select`` (skipped if ``pre_filter_tag``)
2. multi-select — ``questionary.checkbox`` with installed items disabled
if not skills:
console.print("[dim]No skills available.[/dim]")
console.print("[dim]Install with:[/dim] /install-skill <path-or-url>")
console.print(
f"[dim]Global skills:[/dim] [cyan]{_shorten_path(str(GLOBAL_SKILLS_DIR))}[/cyan]"
)
console.print()
return
workspace_skills = [s for s in skills if s.source == "workspace"]
global_skills = [s for s in skills if s.source == "global"]
builtin_skills = [s for s in skills if s.source == "builtin"]
sections = [
("Workspace Skills", workspace_skills, "green"),
("Global Skills", global_skills, "cyan"),
("Built-in Skills", builtin_skills, "blue"),
]
printed = False
for title, group, color in sections:
if not group:
continue
if printed:
console.print()
console.print(f"[bold]{title}[/bold] ({len(group)}):")
for skill in group:
tags_str = f" [dim]({', '.join(skill.tags)})[/dim]" if skill.tags else ""
console.print(
f" [{color}]{skill.name}[/{color}] - {skill.description}{tags_str}"
)
printed = True
console.print(
f"\n[dim]Global skills:[/dim] [cyan]{_shorten_path(str(GLOBAL_SKILLS_DIR))}[/cyan]"
)
console.print(
f"[dim]Workspace skills:[/dim] [green]{_shorten_path(str(USER_SKILLS_DIR))}[/green]"
)
console.print()
def _cmd_install_skill(args: str) -> None:
"""Install a skill from local path or GitHub URL.
By default, installs to the global skills directory (~/.evoscientist/skills/).
Append --local to install to the current workspace instead.
Usage: /install-skill <path-or-url> [--local]
"""
from ..paths import GLOBAL_SKILLS_DIR, USER_SKILLS_DIR
from ..tools.skills_manager import install_skill
# Parse --local flag out of the args string
local = "--local" in args.split()
source = args.replace("--local", "").strip()
if not source:
console.print("[red]Usage:[/red] /install-skill <path-or-url> [--local]")
console.print("[dim]Examples:[/dim]")
console.print(" /install-skill ./my-skill")
console.print(
" /install-skill https://github.com/user/repo/tree/main/skill-name"
)
console.print(" /install-skill user/repo@skill-name")
console.print(
" /install-skill ./my-skill --local [dim](workspace only)[/dim]"
)
console.print()
return
dest_label = (
f"[cyan]{_shorten_path(str(USER_SKILLS_DIR))}[/cyan] [dim](workspace)[/dim]"
if local
else f"[cyan]{_shorten_path(str(GLOBAL_SKILLS_DIR))}[/cyan] [dim](global)[/dim]"
)
console.print(f"[dim]Installing skill from:[/dim] {source}")
console.print(f"[dim]Destination:[/dim] {dest_label}")
result = install_skill(source, global_install=not local)
if result.get("batch"):
# Batch install — multiple skills
for item in result.get("installed", []):
console.print(f"[green]Installed:[/green] {item['name']}")
console.print(
f" [dim]Description:[/dim] {item.get('description', '(none)')}"
)
console.print(
f" [dim]Path:[/dim] [cyan]{_shorten_path(item['path'])}[/cyan]"
)
for item in result.get("failed", []):
console.print(f"[red]Failed:[/red] {item['name']} — {item['error']}")
installed_count = len(result.get("installed", []))
if installed_count:
console.print(f"\n[green]{installed_count} skill(s) installed.[/green]")
console.print("[dim]Reload with /new to apply.[/dim]")
elif result["success"]:
console.print(f"[green]Installed:[/green] {result['name']}")
console.print(f"[dim]Description:[/dim] {result.get('description', '(none)')}")
console.print(f"[dim]Path:[/dim] [cyan]{_shorten_path(result['path'])}[/cyan]")
console.print()
console.print("[dim]Reload with /new to apply.[/dim]")
else:
console.print(f"[red]Failed:[/red] {result['error']}")
console.print()
def _cmd_uninstall_skill(name: str) -> None:
"""Uninstall a user-installed skill."""
from ..tools.skills_manager import uninstall_skill
if not name:
console.print("[red]Usage:[/red] /uninstall-skill <skill-name>")
console.print("[dim]Use /skills to see installed skills.[/dim]")
console.print()
return
result = uninstall_skill(name)
if result["success"]:
console.print(f"[green]Uninstalled:[/green] {name}")
console.print("[dim]Reload with /new to apply.[/dim]")
else:
console.print(f"[red]Failed:[/red] {result['error']}")
console.print()
def _cmd_install_skills(args: str = "") -> None:
"""Browse and install skills from the EvoSkills repository.
Args:
args: Optional tag name to pre-filter (e.g. "core").
Returns:
list of ``install_source`` strings selected by the user,
``None`` if the user cancelled at either phase, or
``[]`` if nothing was selectable / all-installed in the filter.
"""
from collections import Counter
import questionary
from questionary import Choice
from ..paths import GLOBAL_SKILLS_DIR, USER_SKILLS_DIR
from ..tools.skills_manager import fetch_remote_skill_index, install_skill
from .interactive import _PICKER_STYLE
from .widgets.thread_selector import PICKER_STYLE
# Installed-item indicator style for disabled checkbox choices.
_INSTALLED_INDICATOR = ("fg:#4caf50", "✓ ")
@@ -179,46 +49,24 @@ def _cmd_install_skills(args: str = "") -> None:
return questionary.checkbox(
message,
choices=choices,
style=_PICKER_STYLE,
style=PICKER_STYLE,
qmark="❯",
**kwargs,
).ask()
finally:
InquirerControl._get_choice_tokens = original
# Step 1: Fetch remote index
console.print("[dim]Fetching skill index...[/dim]")
try:
index = fetch_remote_skill_index()
except Exception as e:
console.print(f"[red]Failed to fetch skill index: {e}[/red]")
console.print(
"[dim]Try installing directly: /install-skill EvoScientist/EvoSkills@skills[/dim]"
)
console.print()
return
pre_filter_tag = (pre_filter_tag or "").strip().lower()
if not index:
console.print("[yellow]No skills found in the repository.[/yellow]")
console.print()
return
# Detect already-installed skills (both global and workspace tiers)
installed_names: set[str] = set()
for skills_dir in (Path(GLOBAL_SKILLS_DIR), Path(USER_SKILLS_DIR)):
if skills_dir.exists():
installed_names.update(e.name for e in skills_dir.iterdir() if e.is_dir())
pre_filter_tag = args.strip().lower() if args else ""
# Step 2: Tag filter (skip if pre-filtered via args)
# Phase 1: tag filter (skip if pre-filtered via args)
if pre_filter_tag:
filtered = [
s for s in index if pre_filter_tag in [t.lower() for t in s.get("tags", [])]
]
if not filtered:
console.print(f"[yellow]No skills found with tag: {args.strip()}[/yellow]")
# Show available tags
console.print(
f"[yellow]No skills found with tag: {pre_filter_tag}[/yellow]"
)
tag_counter: Counter[str] = Counter()
for s in index:
for t in s.get("tags", []):
@@ -227,10 +75,8 @@ def _cmd_install_skills(args: str = "") -> None:
sorted_tags = sorted(tag_counter.items(), key=lambda x: (-x[1], x[0]))
tags_str = ", ".join(f"{tag} ({count})" for tag, count in sorted_tags)
console.print(f"[dim]Available tags: {tags_str}[/dim]")
console.print()
return
return []
else:
# Build tag choices for interactive picker
tag_counter = Counter()
for s in index:
for t in s.get("tags", []):
@@ -244,13 +90,12 @@ def _cmd_install_skills(args: str = "") -> None:
selected_tag = questionary.select(
"Filter by tag:",
choices=tag_choices,
style=_PICKER_STYLE,
style=PICKER_STYLE,
qmark="❯",
).ask()
if selected_tag is None:
console.print()
return
return None
if selected_tag == "__all__":
filtered = index
@@ -261,14 +106,12 @@ def _cmd_install_skills(args: str = "") -> None:
if selected_tag in [t.lower() for t in s.get("tags", [])]
]
# Step 3: Skill selection checkbox
all_installed = all(s["name"] in installed_names for s in filtered)
if all_installed:
# Phase 2: skill selection checkbox
if all(s["name"] in installed_names for s in filtered):
console.print(
"[green]All skills in this category are already installed.[/green]"
)
console.print()
return
return []
choices = []
for s in filtered:
@@ -294,31 +137,5 @@ def _cmd_install_skills(args: str = "") -> None:
selected = _checkbox_ask(choices, "Select skills to install:")
if selected is None:
console.print()
return
if not selected:
console.print("[dim]No skills selected.[/dim]")
console.print()
return
# Step 4: Install selected skills (default: global)
installed_count = 0
for source in selected:
result = install_skill(source, global_install=True)
if result.get("batch"):
for item in result.get("installed", []):
console.print(f"[green]Installed:[/green] {item['name']}")
installed_count += 1
for item in result.get("failed", []):
console.print(f"[red]Failed:[/red] {item['name']} — {item['error']}")
elif result.get("success"):
console.print(f"[green]Installed:[/green] {result['name']}")
installed_count += 1
else:
console.print(f"[red]Failed:[/red] {result.get('error', 'unknown')}")
if installed_count:
console.print(f"\n[green]{installed_count} skill(s) installed.[/green]")
console.print("[dim]Reload with /new to apply.[/dim]")
console.print()
return None
return list(selected)
EvoScientist/cli/widgets/thread_selector.py
+17
View File
@@ -18,6 +18,7 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Any, ClassVar
from prompt_toolkit.styles import Style as PtStyle # type: ignore[import-untyped]
from rich.text import Text
from textual.binding import Binding, BindingType
from textual.containers import Container
@@ -30,6 +31,22 @@ if TYPE_CHECKING:
from textual.app import ComposeResult
# Style for questionary pickers used by the Rich CLI ``/resume`` and
# ``/delete`` interactive selectors. Matches the slash-completion menu's
# visual language: gray (#888888) for non-selected, bold for selected,
# no background changes.
PICKER_STYLE = PtStyle.from_dict(
{
"questionmark": "#888888",
"question": "",
"pointer": "bold",
"highlighted": "bold",
"text": "#888888",
"answer": "bold",
}
)
# ---------------------------------------------------------------------------
# Path helpers
# ---------------------------------------------------------------------------
EvoScientist/commands/implementation/channel.py
+12
View File
@@ -98,6 +98,12 @@ class ChannelCommand(Command):
ctx.ui.append_system(f"Adding channel(s): {', '.join(requested)}...")
from ...cli.channel import _add_channel_to_running_bus
# Sync CLI-side globals up-front so partial-success states (one
# channel attached, next one raises) still leave
# _auto_start_channel observing the latest agent/thread refs.
# See cli/channel.py _cli_agent / _cli_thread_id.
_ch_mod._cli_agent = ctx.agent
_ch_mod._cli_thread_id = ctx.thread_id
try:
for ct in requested:
_add_channel_to_running_bus(ct, config, send_thinking=send_thinking)
@@ -131,6 +137,12 @@ class ChannelCommand(Command):
ctx.thread_id,
send_thinking=send_thinking,
)
# Sync CLI-side globals so _auto_start_channel observes the
# latest agent/thread refs (see cli/channel.py _cli_agent /
# _cli_thread_id). The pre-migration inline helper set these
# in the same branch.
_ch_mod._cli_agent = ctx.agent
_ch_mod._cli_thread_id = ctx.thread_id
# Show status panel
if _ch_mod._manager:
EvoScientist/commands/implementation/session.py
+5
View File
@@ -113,6 +113,11 @@ class ThreadsCommand(Command):
table.add_row(*row)
ctx.ui.mount_renderable(table)
if not is_channel:
ctx.ui.append_system(
" /resume to continue a session "
"/delete <id> to remove /new to start fresh",
)
class ResumeCommand(Command):
EvoScientist/commands/implementation/skills.py
+8 -1
View File
@@ -213,11 +213,18 @@ class InstallSkills(Command):
pre_filter_tag=tag,
)
if not selected_sources:
# ``None`` means user cancelled (Esc / Ctrl-C). An empty list means
# the picker handled a "nothing to do" state (all-installed / no
# tag matches) and already printed its own specific message; the
# outer layer should stay silent rather than claim a cancel.
if selected_sources is None:
if not is_channel:
ctx.ui.append_system("Browse cancelled.", style="dim")
return
if not selected_sources:
return
# Install selected skills
installed_count = 0
for source in selected_sources:
EvoScientist/stream/display.py
+1 -1
View File
@@ -989,7 +989,7 @@ def _resolve_ask_user_prompt(ask_user_data: dict) -> dict:
"""
import questionary # type: ignore[import-untyped]
from ..cli.interactive import _PICKER_STYLE
from ..cli.widgets.thread_selector import PICKER_STYLE as _PICKER_STYLE
questions = ask_user_data.get("questions", [])
if not questions:
tests/test_channel_command.py
+219
View File
@@ -0,0 +1,219 @@
"""Tests for the /channel command (ChannelCommand)."""
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
from tests.conftest import run_async as _run
@pytest.fixture(autouse=True)
def _reset_channel_globals():
"""Reset module-level ``_cli_agent``/``_cli_thread_id`` around every test.
These globals are written by the start / add-to-running paths and would
otherwise leak between tests (and into unrelated suites).
"""
import EvoScientist.cli.channel as _ch
_ch._cli_agent = None
_ch._cli_thread_id = None
try:
yield
finally:
_ch._cli_agent = None
_ch._cli_thread_id = None
def _ctx():
from EvoScientist.commands.base import CommandContext
ui = MagicMock()
ui.supports_interactive = True
return CommandContext(
agent=object(), thread_id="tid-42", ui=ui, workspace_dir="/ws"
), ui
class TestNeedsAgent:
"""status + stop must not require the agent (recovery from broken load)."""
def test_status_does_not_need_agent(self):
from EvoScientist.commands.implementation.channel import ChannelCommand
assert ChannelCommand().needs_agent(["status"]) is False
def test_stop_does_not_need_agent(self):
from EvoScientist.commands.implementation.channel import ChannelCommand
assert ChannelCommand().needs_agent(["stop"]) is False
def test_stop_with_target_does_not_need_agent(self):
from EvoScientist.commands.implementation.channel import ChannelCommand
assert ChannelCommand().needs_agent(["stop", "telegram"]) is False
def test_start_subcommand_needs_agent(self):
from EvoScientist.commands.implementation.channel import ChannelCommand
assert ChannelCommand().needs_agent(["telegram"]) is True
def test_no_args_needs_agent(self):
from EvoScientist.commands.implementation.channel import ChannelCommand
# no subcmd → falls through to start flow, needs agent
assert ChannelCommand().needs_agent([]) is True
class TestStartPath:
"""Start flow must propagate agent/thread_id globals."""
def test_start_sets_cli_agent_globals(self):
import EvoScientist.cli.channel as _ch_mod
from EvoScientist.commands.implementation.channel import ChannelCommand
ctx, _ui = _ctx()
config = SimpleNamespace(
channel_enabled="telegram",
channel_send_thinking=True,
)
with (
patch(
"EvoScientist.cli.channel._channels_is_running",
return_value=False,
),
patch(
"EvoScientist.cli.channel._start_channels_bus_mode",
),
patch(
"EvoScientist.config.load_config",
return_value=config,
),
):
_run(ChannelCommand().execute(ctx, ["telegram"]))
assert _ch_mod._cli_agent is ctx.agent
assert _ch_mod._cli_thread_id == "tid-42"
def test_start_propagates_send_thinking(self):
"""send_thinking flag must reach _start_channels_bus_mode."""
from EvoScientist.commands.implementation.channel import ChannelCommand
ctx, _ui = _ctx()
config = SimpleNamespace(
channel_enabled="telegram",
channel_send_thinking=False,
)
captured = {}
def _fake_start(cfg, agent, thread_id, *, send_thinking=None):
captured["agent"] = agent
captured["thread_id"] = thread_id
captured["send_thinking"] = send_thinking
with (
patch(
"EvoScientist.cli.channel._channels_is_running",
return_value=False,
),
patch(
"EvoScientist.cli.channel._start_channels_bus_mode",
_fake_start,
),
patch(
"EvoScientist.config.load_config",
return_value=config,
),
):
_run(ChannelCommand().execute(ctx, ["telegram"]))
assert captured["agent"] is ctx.agent
assert captured["thread_id"] == "tid-42"
assert captured["send_thinking"] is False
class TestAddToRunningPath:
def test_add_to_running_sets_cli_agent_globals(self):
import EvoScientist.cli.channel as _ch_mod
from EvoScientist.commands.implementation.channel import ChannelCommand
ctx, _ui = _ctx()
config = SimpleNamespace(
channel_enabled="telegram",
channel_send_thinking=False,
)
with (
patch(
"EvoScientist.cli.channel._channels_is_running",
return_value=True,
),
patch(
"EvoScientist.cli.channel._add_channel_to_running_bus",
),
patch(
"EvoScientist.config.load_config",
return_value=config,
),
):
_run(ChannelCommand().execute(ctx, ["discord"]))
assert _ch_mod._cli_agent is ctx.agent
assert _ch_mod._cli_thread_id == "tid-42"
def test_add_to_running_propagates_send_thinking(self):
"""Adding to a running bus must honor config.channel_send_thinking."""
from EvoScientist.commands.implementation.channel import ChannelCommand
ctx, _ui = _ctx()
config = SimpleNamespace(
channel_enabled="telegram",
channel_send_thinking=True,
)
captured = {}
def _fake_add(channel_type, cfg, *, send_thinking=None):
captured["channel_type"] = channel_type
captured["send_thinking"] = send_thinking
with (
patch(
"EvoScientist.cli.channel._channels_is_running",
return_value=True,
),
patch(
"EvoScientist.cli.channel._add_channel_to_running_bus",
_fake_add,
),
patch(
"EvoScientist.config.load_config",
return_value=config,
),
):
_run(ChannelCommand().execute(ctx, ["discord"]))
assert captured["channel_type"] == "discord"
assert captured["send_thinking"] is True
class TestStatusPath:
def test_status_without_running_channels(self):
from EvoScientist.commands.implementation.channel import ChannelCommand
ctx, ui = _ctx()
config = SimpleNamespace(channel_enabled="", channel_send_thinking=False)
with (
patch(
"EvoScientist.cli.channel._channels_is_running",
return_value=False,
),
patch(
"EvoScientist.cli.channel._channels_running_list",
return_value=[],
),
patch(
"EvoScientist.config.load_config",
return_value=config,
),
):
_run(ChannelCommand().execute(ctx, ["status"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("No messaging channels" in m for m in msgs)
tests/test_cli_channel_bus_mode.py
-74
View File
@@ -50,77 +50,3 @@ def test_auto_start_channel_passes_send_thinking(monkeypatch):
assert captured["send_thinking"] is False
assert captured["thread_id"] == "thread-1"
assert captured["agent"] is agent
def test_cmd_channel_running_path_passes_send_thinking(monkeypatch):
import EvoScientist.config as config_mod
captured = {}
config = SimpleNamespace(channel_enabled="telegram")
monkeypatch.setattr(config_mod, "load_config", lambda: config)
monkeypatch.setattr(
channel_cli, "_channels_is_running", lambda _channel_type=None: True
)
monkeypatch.setattr(channel_cli, "_channels_running_list", lambda: [])
monkeypatch.setattr(channel_cli, "_print_channel_panel", lambda _rows: None)
monkeypatch.setattr(
channel_cli,
"_add_channel_to_running_bus",
lambda channel_type, cfg, *, send_thinking=None: captured.update(
{
"channel_type": channel_type,
"config": cfg,
"send_thinking": send_thinking,
}
),
)
channel_cli._cmd_channel(
"telegram",
object(),
"thread-1",
send_thinking=False,
)
assert captured["channel_type"] == "telegram"
assert captured["config"] is config
assert captured["send_thinking"] is False
def test_cmd_channel_start_path_passes_send_thinking(monkeypatch):
import EvoScientist.config as config_mod
captured = {}
config = SimpleNamespace(channel_enabled="telegram")
monkeypatch.setattr(config_mod, "load_config", lambda: config)
monkeypatch.setattr(
channel_cli, "_channels_is_running", lambda _channel_type=None: False
)
monkeypatch.setattr(channel_cli, "_print_channel_panel", lambda _rows: None)
monkeypatch.setattr(
channel_cli,
"_start_channels_bus_mode",
lambda cfg, agent, thread_id, *, send_thinking=None: captured.update(
{
"config": cfg,
"agent": agent,
"thread_id": thread_id,
"send_thinking": send_thinking,
}
),
)
agent = object()
channel_cli._cmd_channel(
"",
agent,
"thread-1",
send_thinking=False,
)
assert captured["config"] is config
assert captured["agent"] is agent
assert captured["thread_id"] == "thread-1"
assert captured["send_thinking"] is False
tests/test_current_command.py
+41
View File
@@ -0,0 +1,41 @@
"""Tests for the /current command."""
from unittest.mock import MagicMock
from tests.conftest import run_async as _run
class TestCurrentCommand:
def test_prints_thread_workspace_and_memory(self):
from EvoScientist.commands.base import CommandContext
from EvoScientist.commands.implementation.general import CurrentCommand
ui = MagicMock()
ctx = CommandContext(
agent=None,
thread_id="abc123",
ui=ui,
workspace_dir="/tmp/ws",
)
_run(CurrentCommand().execute(ctx, []))
# Three append_system calls: Thread, Workspace, Memory dir.
calls = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Thread: abc123" in s for s in calls)
assert any("Workspace:" in s for s in calls)
assert any("Memory dir:" in s for s in calls)
def test_skips_workspace_when_none(self):
from EvoScientist.commands.base import CommandContext
from EvoScientist.commands.implementation.general import CurrentCommand
ui = MagicMock()
ctx = CommandContext(
agent=None,
thread_id="abc123",
ui=ui,
workspace_dir=None,
)
_run(CurrentCommand().execute(ctx, []))
calls = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Thread: abc123" in s for s in calls)
assert not any("Workspace:" in s for s in calls)
tests/test_delete_command.py
+141
View File
@@ -0,0 +1,141 @@
"""Tests for the /delete command."""
from unittest.mock import AsyncMock, MagicMock, patch
from tests.conftest import run_async as _run
def _ctx(thread_id="current"):
from EvoScientist.commands.base import CommandContext
ui = MagicMock()
ui.supports_interactive = True
return CommandContext(agent=None, thread_id=thread_id, ui=ui), ui
def _patches(thread_exists=False, similar=None, deleted=True, threads=None):
"""Return a context manager stack patching the sessions module."""
from contextlib import ExitStack
stack = ExitStack()
stack.enter_context(
patch(
"EvoScientist.sessions.thread_exists",
new=AsyncMock(return_value=thread_exists),
)
)
stack.enter_context(
patch(
"EvoScientist.sessions.find_similar_threads",
new=AsyncMock(return_value=similar or []),
)
)
stack.enter_context(
patch(
"EvoScientist.sessions.delete_thread",
new=AsyncMock(return_value=deleted),
)
)
stack.enter_context(
patch(
"EvoScientist.sessions.list_threads",
new=AsyncMock(return_value=threads or []),
)
)
return stack
class TestDeleteCommand:
def test_refuses_to_delete_current(self):
from EvoScientist.commands.implementation.session import DeleteCommand
ctx, ui = _ctx(thread_id="current")
# Inline the patches here (rather than using ``_patches``) so we
# can keep a direct handle on the ``delete_thread`` mock and
# assert on it *inside* the context. Asserting after the
# context exits hits the real function (no ``await_count``
# attr), which silently degrades into ``assert True``.
mock_delete = AsyncMock(return_value=True)
with (
patch(
"EvoScientist.sessions.thread_exists",
new=AsyncMock(return_value=True),
),
patch("EvoScientist.sessions.delete_thread", new=mock_delete),
patch(
"EvoScientist.sessions.find_similar_threads",
new=AsyncMock(return_value=[]),
),
patch(
"EvoScientist.sessions.list_threads",
new=AsyncMock(return_value=[]),
),
):
_run(DeleteCommand().execute(ctx, ["current"]))
assert mock_delete.await_count == 0
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Cannot delete the current session" in m for m in msgs)
def test_happy_path_success(self):
from EvoScientist.commands.implementation.session import DeleteCommand
ctx, ui = _ctx(thread_id="current")
with _patches(thread_exists=True, deleted=True):
_run(DeleteCommand().execute(ctx, ["other-thread"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Deleted session other-thread" in m for m in msgs)
def test_not_found(self):
from EvoScientist.commands.implementation.session import DeleteCommand
ctx, ui = _ctx()
with _patches(thread_exists=False, similar=[]):
_run(DeleteCommand().execute(ctx, ["missing"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("not found" in m for m in msgs)
def test_ambiguous_prefix(self):
from EvoScientist.commands.implementation.session import DeleteCommand
ctx, ui = _ctx()
with _patches(thread_exists=False, similar=["abc-one", "abc-two"]):
_run(DeleteCommand().execute(ctx, ["abc"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Ambiguous" in m for m in msgs)
def test_prefix_resolves_to_unique_match(self):
from EvoScientist.commands.implementation.session import DeleteCommand
ctx, ui = _ctx()
with _patches(thread_exists=False, similar=["abc-one"], deleted=True):
_run(DeleteCommand().execute(ctx, ["abc"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Deleted session abc-one" in m for m in msgs)
def test_no_arg_empty_sessions_prints_notice(self):
from EvoScientist.commands.implementation.session import DeleteCommand
ctx, ui = _ctx()
with _patches(threads=[]):
_run(DeleteCommand().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("No sessions to delete" in m for m in msgs)
def test_no_arg_calls_picker_returns_none(self):
"""When no arg and picker returns None, nothing is deleted."""
from EvoScientist.commands.implementation.session import DeleteCommand
ctx, ui = _ctx()
ui.wait_for_thread_pick = AsyncMock(return_value=None)
threads = [
{
"thread_id": "t1",
"preview": "",
"message_count": 1,
"model": None,
"updated_at": None,
}
]
with _patches(threads=threads):
_run(DeleteCommand().execute(ctx, []))
ui.wait_for_thread_pick.assert_awaited_once()
tests/test_evoskills_command.py
+104
View File
@@ -0,0 +1,104 @@
"""Tests for the /evoskills command (InstallSkills)."""
from unittest.mock import AsyncMock, MagicMock, patch
from tests.conftest import run_async as _run
def _ctx(supports_interactive=True):
from EvoScientist.commands.base import CommandContext
ui = MagicMock()
ui.supports_interactive = supports_interactive
ui.wait_for_skill_browse = AsyncMock()
return CommandContext(agent=None, thread_id="tid", ui=ui), ui
_INDEX = [
{
"name": "paper-writing",
"description": "author papers",
"install_source": "repo@paper-writing",
"tags": ["writing"],
},
{
"name": "research-ideation",
"description": "brainstorm ideas",
"install_source": "repo@research-ideation",
"tags": ["core"],
},
]
class TestInstallSkills:
def test_picker_cancel_no_install(self):
from EvoScientist.commands.implementation.skills import InstallSkills
ctx, ui = _ctx()
ui.wait_for_skill_browse.return_value = None # cancelled
with (
patch(
"EvoScientist.tools.skills_manager.fetch_remote_skill_index",
return_value=_INDEX,
),
patch(
"EvoScientist.tools.skills_manager.install_skill",
) as install_mock,
):
_run(InstallSkills().execute(ctx, []))
install_mock.assert_not_called()
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Browse cancelled" in m for m in msgs)
def test_picker_returns_selections_installs_each(self):
from EvoScientist.commands.implementation.skills import InstallSkills
ctx, ui = _ctx()
ui.wait_for_skill_browse.return_value = [
"repo@paper-writing",
"repo@research-ideation",
]
with (
patch(
"EvoScientist.tools.skills_manager.fetch_remote_skill_index",
return_value=_INDEX,
),
patch(
"EvoScientist.tools.skills_manager.install_skill",
return_value={"success": True, "name": "x"},
) as install_mock,
):
_run(InstallSkills().execute(ctx, []))
assert install_mock.call_count == 2
def test_channel_auto_install_on_tag(self):
"""Non-interactive UI + tag arg → auto-installs matching skills."""
from EvoScientist.commands.implementation.skills import InstallSkills
ctx, ui = _ctx(supports_interactive=False)
with (
patch(
"EvoScientist.tools.skills_manager.fetch_remote_skill_index",
return_value=_INDEX,
),
patch(
"EvoScientist.tools.skills_manager.install_skill",
return_value={"success": True, "name": "x"},
) as install_mock,
):
_run(InstallSkills().execute(ctx, ["core"]))
# "core" matches research-ideation only → 1 install, no picker call
assert install_mock.call_count == 1
ui.wait_for_skill_browse.assert_not_called()
def test_fetch_failure_prints_error(self):
from EvoScientist.commands.implementation.skills import InstallSkills
ctx, ui = _ctx()
with patch(
"EvoScientist.tools.skills_manager.fetch_remote_skill_index",
side_effect=RuntimeError("network fail"),
):
_run(InstallSkills().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Failed to fetch" in m for m in msgs)
tests/test_exit_command.py
+29
View File
@@ -0,0 +1,29 @@
"""Tests for the /exit command (ExitCommand → ctx.ui.force_quit)."""
from unittest.mock import MagicMock
from tests.conftest import run_async as _run
class TestExitCommand:
def test_execute_calls_force_quit(self):
from EvoScientist.commands.base import CommandContext
from EvoScientist.commands.implementation.session import ExitCommand
ui = MagicMock()
ctx = CommandContext(
agent=None,
thread_id="tid",
ui=ui,
)
cmd = ExitCommand()
_run(cmd.execute(ctx, []))
ui.force_quit.assert_called_once()
def test_aliases_registered(self):
"""/quit and /q resolve to the same ExitCommand as /exit."""
from EvoScientist.commands.manager import manager
assert manager.get_command("/exit").name == "/exit"
assert manager.get_command("/quit").name == "/exit"
assert manager.get_command("/q").name == "/exit"
tests/test_install_skill_command.py
+74
View File
@@ -0,0 +1,74 @@
"""Tests for /install-skill and /uninstall-skill commands."""
from unittest.mock import MagicMock, patch
from tests.conftest import run_async as _run
def _ctx():
from EvoScientist.commands.base import CommandContext
ui = MagicMock()
ui.supports_interactive = True
return CommandContext(agent=None, thread_id="tid", ui=ui), ui
class TestInstallSkill:
def test_usage_message_when_no_args(self):
from EvoScientist.commands.implementation.skills import InstallSkill
ctx, ui = _ctx()
_run(InstallSkill().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Usage:" in m for m in msgs)
def test_happy_path(self):
from EvoScientist.commands.implementation.skills import InstallSkill
ctx, ui = _ctx()
with patch(
"EvoScientist.tools.skills_manager.install_skill",
return_value={
"success": True,
"name": "demo-skill",
"description": "demo",
"path": "/tmp/demo",
},
):
_run(InstallSkill().execute(ctx, ["./some-path"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Installed: demo-skill" in m for m in msgs)
class TestUninstallSkill:
def test_usage_message_when_no_args(self):
from EvoScientist.commands.implementation.skills import UninstallSkill
ctx, ui = _ctx()
_run(UninstallSkill().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Usage:" in m for m in msgs)
def test_uninstall_success(self):
from EvoScientist.commands.implementation.skills import UninstallSkill
ctx, ui = _ctx()
with patch(
"EvoScientist.tools.skills_manager.uninstall_skill",
return_value={"success": True},
):
_run(UninstallSkill().execute(ctx, ["demo-skill"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Uninstalled: demo-skill" in m for m in msgs)
def test_uninstall_failure(self):
from EvoScientist.commands.implementation.skills import UninstallSkill
ctx, ui = _ctx()
with patch(
"EvoScientist.tools.skills_manager.uninstall_skill",
return_value={"success": False, "error": "not found"},
):
_run(UninstallSkill().execute(ctx, ["missing"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Failed: not found" in m for m in msgs)
tests/test_mcp_command.py
+113
View File
@@ -0,0 +1,113 @@
"""Tests for the /mcp command (MCPCommand subcommand dispatch)."""
from unittest.mock import MagicMock, patch
from tests.conftest import run_async as _run
def _ctx():
from EvoScientist.commands.base import CommandContext
ui = MagicMock()
ui.supports_interactive = True
return CommandContext(agent=None, thread_id="tid", ui=ui), ui
class TestMCPCommandDispatch:
def test_no_args_lists(self):
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, ui = _ctx()
with patch("EvoScientist.mcp.load_mcp_config", return_value={}):
_run(MCPCommand().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("No MCP servers configured" in m for m in msgs)
def test_list_subcommand(self):
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, ui = _ctx()
cfg = {
"srv1": {"transport": "stdio", "tools": ["foo"], "expose_to": ["main"]},
}
with patch("EvoScientist.mcp.load_mcp_config", return_value=cfg):
_run(MCPCommand().execute(ctx, ["list"]))
ui.mount_renderable.assert_called_once()
def test_add_subcommand_dispatches(self):
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, _ui = _ctx()
with (
patch(
"EvoScientist.mcp.parse_mcp_add_args",
return_value={"name": "srv1"},
),
patch(
"EvoScientist.mcp.add_mcp_server",
return_value={"transport": "stdio"},
) as add_mock,
):
_run(MCPCommand().execute(ctx, ["add", "srv1", "python"]))
add_mock.assert_called_once()
def test_edit_subcommand_dispatches(self):
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, _ui = _ctx()
with (
patch(
"EvoScientist.mcp.parse_mcp_edit_args",
return_value=("srv1", {"tools": ["bar"]}),
),
patch(
"EvoScientist.mcp.edit_mcp_server",
) as edit_mock,
):
_run(MCPCommand().execute(ctx, ["edit", "srv1", "--tools", "bar"]))
edit_mock.assert_called_once_with("srv1", tools=["bar"])
def test_remove_subcommand_success(self):
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, ui = _ctx()
with patch("EvoScientist.mcp.remove_mcp_server", return_value=True):
_run(MCPCommand().execute(ctx, ["remove", "srv1"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Removed MCP server: srv1" in m for m in msgs)
def test_remove_subcommand_not_found(self):
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, ui = _ctx()
with patch("EvoScientist.mcp.remove_mcp_server", return_value=False):
_run(MCPCommand().execute(ctx, ["remove", "missing"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Server not found" in m for m in msgs)
def test_install_delegates_to_install_mcp_command(self):
"""/mcp install should instantiate InstallMCPCommand and execute it."""
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, _ui = _ctx()
with patch(
"EvoScientist.commands.implementation.mcp_install.InstallMCPCommand"
) as klass:
instance = MagicMock()
instance.execute = MagicMock(return_value=None)
async def fake_execute(ctx, args):
return None
instance.execute = fake_execute
klass.return_value = instance
_run(MCPCommand().execute(ctx, ["install", "foo"]))
klass.assert_called_once()
def test_unknown_subcommand_prints_help(self):
from EvoScientist.commands.implementation.mcp import MCPCommand
ctx, ui = _ctx()
_run(MCPCommand().execute(ctx, ["bogus"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("MCP commands:" in m for m in msgs)
tests/test_new_command.py
+36
View File
@@ -0,0 +1,36 @@
"""Tests for the /new command."""
from unittest.mock import MagicMock
from tests.conftest import run_async as _run
class TestNewCommand:
def test_execute_calls_start_new_session(self):
from EvoScientist.commands.base import CommandContext
from EvoScientist.commands.implementation.session import NewCommand
ui = MagicMock()
ctx = CommandContext(
agent=None,
thread_id="old-tid",
ui=ui,
workspace_dir="/old/ws",
)
_run(NewCommand().execute(ctx, []))
ui.start_new_session.assert_called_once()
def test_requires_agent_false(self):
from EvoScientist.commands.implementation.session import NewCommand
assert NewCommand().requires_agent is False
def test_no_agent_access(self):
"""Command body must not touch ctx.agent (it's still loading)."""
from EvoScientist.commands.base import CommandContext
from EvoScientist.commands.implementation.session import NewCommand
ui = MagicMock()
ctx = CommandContext(agent=None, thread_id="tid", ui=ui)
# No AttributeError even though ctx.agent is None
_run(NewCommand().execute(ctx, []))
tests/test_pick_skills_interactive.py
+125
View File
@@ -0,0 +1,125 @@
"""Tests for _pick_skills_interactive (Phase C extraction).
Verifies empty-result vs cancel semantics so InstallSkills.execute
can distinguish "picker handled the no-op case" from "user cancelled".
"""
from unittest.mock import MagicMock, patch
_INDEX = [
{
"name": "paper-writing",
"description": "author papers",
"install_source": "repo@paper-writing",
"tags": ["writing"],
},
{
"name": "research-ideation",
"description": "brainstorm ideas",
"install_source": "repo@research-ideation",
"tags": ["core"],
},
]
class TestPickSkillsInteractive:
def test_no_tag_match_returns_empty_list(self):
"""Pre-filter with no matches → [] (not None). Caller should
suppress its own "cancelled" message since the picker already
printed a specific one."""
from EvoScientist.cli.skills_cmd import _pick_skills_interactive
result = _pick_skills_interactive(_INDEX, set(), "nonexistent-tag")
assert result == []
def test_all_installed_returns_empty_list(self):
"""If every skill matching the tag is already installed → []."""
from EvoScientist.cli.skills_cmd import _pick_skills_interactive
installed = {"paper-writing", "research-ideation"}
# Pre-filter skips tag picker; directly hits all-installed guard
result = _pick_skills_interactive(_INDEX, installed, "writing")
assert result == []
def test_tag_picker_cancel_returns_none(self, monkeypatch):
"""User cancels tag picker (Esc) → None."""
from EvoScientist.cli import skills_cmd
select_prompt = MagicMock()
select_prompt.ask.return_value = None
monkeypatch.setattr("questionary.select", lambda *a, **k: select_prompt)
result = skills_cmd._pick_skills_interactive(_INDEX, set(), "")
assert result is None
def test_checkbox_cancel_returns_none(self, monkeypatch):
"""User cancels checkbox (Esc) → None."""
from EvoScientist.cli import skills_cmd
# Skip tag picker by pre-filtering
checkbox_prompt = MagicMock()
checkbox_prompt.ask.return_value = None
monkeypatch.setattr("questionary.checkbox", lambda *a, **k: checkbox_prompt)
result = skills_cmd._pick_skills_interactive(_INDEX, set(), "writing")
assert result is None
def test_checkbox_confirmed_with_selection(self, monkeypatch):
"""User confirms with selections → list of install sources."""
from EvoScientist.cli import skills_cmd
checkbox_prompt = MagicMock()
checkbox_prompt.ask.return_value = ["repo@paper-writing"]
monkeypatch.setattr("questionary.checkbox", lambda *a, **k: checkbox_prompt)
result = skills_cmd._pick_skills_interactive(_INDEX, set(), "writing")
assert result == ["repo@paper-writing"]
class TestInstallSkillsHandlesEmpty:
"""InstallSkills.execute must distinguish None vs [] from the picker."""
def test_empty_list_suppresses_cancel_message(self):
"""When picker returns [], user should NOT see "Browse cancelled"
(the picker already printed its own message)."""
from unittest.mock import AsyncMock
from EvoScientist.commands.base import CommandContext
from EvoScientist.commands.implementation.skills import InstallSkills
from tests.conftest import run_async as _run
ui = MagicMock()
ui.supports_interactive = True
ui.wait_for_skill_browse = AsyncMock(return_value=[])
ctx = CommandContext(agent=None, thread_id="tid", ui=ui)
with patch(
"EvoScientist.tools.skills_manager.fetch_remote_skill_index",
return_value=_INDEX,
):
_run(InstallSkills().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert not any("Browse cancelled" in m for m in msgs)
def test_none_shows_cancel_message(self):
"""When picker returns None (actual cancel), user sees the message."""
from unittest.mock import AsyncMock
from EvoScientist.commands.base import CommandContext
from EvoScientist.commands.implementation.skills import InstallSkills
from tests.conftest import run_async as _run
ui = MagicMock()
ui.supports_interactive = True
ui.wait_for_skill_browse = AsyncMock(return_value=None)
ctx = CommandContext(agent=None, thread_id="tid", ui=ui)
with patch(
"EvoScientist.tools.skills_manager.fetch_remote_skill_index",
return_value=_INDEX,
):
_run(InstallSkills().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Browse cancelled" in m for m in msgs)
tests/test_resume_command.py
+142
View File
@@ -0,0 +1,142 @@
"""Tests for the /resume command."""
from contextlib import ExitStack
from unittest.mock import AsyncMock, MagicMock, patch
from tests.conftest import run_async as _run
def _ctx(thread_id="current", workspace_dir="/ws"):
from EvoScientist.commands.base import CommandContext
ui = MagicMock()
ui.supports_interactive = True
ui.wait_for_thread_pick = AsyncMock()
ui.handle_session_resume = AsyncMock()
return CommandContext(
agent=None, thread_id=thread_id, ui=ui, workspace_dir=workspace_dir
), ui
def _patches(
*,
thread_exists=False,
similar=None,
threads=None,
metadata=None,
):
stack = ExitStack()
stack.enter_context(
patch(
"EvoScientist.sessions.thread_exists",
new=AsyncMock(return_value=thread_exists),
)
)
stack.enter_context(
patch(
"EvoScientist.sessions.find_similar_threads",
new=AsyncMock(return_value=similar or []),
)
)
stack.enter_context(
patch(
"EvoScientist.sessions.list_threads",
new=AsyncMock(return_value=threads or []),
)
)
stack.enter_context(
patch(
"EvoScientist.sessions.get_thread_metadata",
new=AsyncMock(return_value=metadata or {}),
)
)
return stack
class TestResumeCommand:
def test_with_arg_resolves_and_calls_ui(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx()
with _patches(thread_exists=True, metadata={"workspace_dir": "/restored"}):
_run(ResumeCommand().execute(ctx, ["target-tid"]))
ui.handle_session_resume.assert_awaited_once_with("target-tid", "/restored")
# ctx mutations
assert ctx.thread_id == "target-tid"
assert ctx.workspace_dir == "/restored"
def test_no_arg_empty_threads_prints_message(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx()
with _patches(threads=[]):
_run(ResumeCommand().execute(ctx, []))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("No sessions to resume" in m for m in msgs)
ui.wait_for_thread_pick.assert_not_called()
ui.handle_session_resume.assert_not_called()
def test_no_arg_calls_picker(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx()
ui.wait_for_thread_pick.return_value = "picked-tid"
threads = [{"thread_id": "picked-tid", "preview": "p", "message_count": 1}]
with _patches(thread_exists=True, threads=threads):
_run(ResumeCommand().execute(ctx, []))
ui.wait_for_thread_pick.assert_awaited_once()
ui.handle_session_resume.assert_awaited_once()
def test_picker_cancel_returns(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx()
ui.wait_for_thread_pick.return_value = None
threads = [{"thread_id": "t1", "preview": "", "message_count": 0}]
with _patches(threads=threads):
_run(ResumeCommand().execute(ctx, []))
ui.handle_session_resume.assert_not_called()
def test_ambiguous_prefix(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx()
with _patches(thread_exists=False, similar=["abc-one", "abc-two"]):
_run(ResumeCommand().execute(ctx, ["abc"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("Ambiguous" in m for m in msgs)
ui.handle_session_resume.assert_not_called()
def test_not_found(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx()
with _patches(thread_exists=False, similar=[]):
_run(ResumeCommand().execute(ctx, ["missing"]))
msgs = [c.args[0] for c in ui.append_system.call_args_list]
assert any("not found" in m for m in msgs)
ui.handle_session_resume.assert_not_called()
def test_prefix_resolves_to_unique_match(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx()
with _patches(
thread_exists=False,
similar=["abc-one"],
metadata={"workspace_dir": "/ws1"},
):
_run(ResumeCommand().execute(ctx, ["abc"]))
ui.handle_session_resume.assert_awaited_once_with("abc-one", "/ws1")
assert ctx.thread_id == "abc-one"
def test_empty_workspace_metadata_preserves_ctx_workspace(self):
from EvoScientist.commands.implementation.session import ResumeCommand
ctx, ui = _ctx(workspace_dir="/keep")
with _patches(thread_exists=True, metadata={}):
_run(ResumeCommand().execute(ctx, ["tid"]))
# ResumeCommand only overwrites ctx.workspace_dir if metadata has one
assert ctx.workspace_dir == "/keep"
# Callback still fires with the metadata value (empty string)
ui.handle_session_resume.assert_awaited_once_with("tid", "")
tests/test_rich_command_ui.py
+234 -39
View File
@@ -2,19 +2,18 @@
from unittest.mock import MagicMock
import pytest
from rich.console import Console
from rich.table import Table
from tests.conftest import run_async as _run
def _make_ui():
def _make_ui(**kwargs):
"""Build a RichCLICommandUI backed by a MagicMock console."""
from EvoScientist.cli.rich_command_ui import RichCLICommandUI
console = MagicMock(spec=Console)
ui = RichCLICommandUI(console)
ui = RichCLICommandUI(console, **kwargs)
return ui, console
@@ -125,50 +124,246 @@ class TestUpdateStatusHook:
console.print.assert_not_called()
class TestUnmigratedMethodsStub:
"""Protocol methods not yet wired for CLI must raise NotImplementedError.
class TestPhaseAMigrated:
"""Phase A migration: quit, clear, thread-pick fallback, and compact status.
These stubs are signposts for the A1 migration (see
cli-commandmanager-migration.md). Each one is replaced by a real
implementation when its corresponding command is migrated.
These replaced the original ``NotImplementedError`` stubs once the
corresponding commands were migrated through the shared CommandManager
dispatch block in ``interactive.py``.
"""
def test_wait_for_thread_pick(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/threads"):
_run(ui.wait_for_thread_pick([], "tid", "title"))
def test_request_quit_fires_callback(self):
called = []
ui, _ = _make_ui(on_request_quit=lambda: called.append("q"))
ui.request_quit()
assert called == ["q"]
def test_wait_for_skill_browse(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/skills"):
_run(ui.wait_for_skill_browse([], set(), ""))
def test_request_quit_without_callback_is_noop(self):
ui, console = _make_ui()
ui.request_quit()
console.clear.assert_not_called()
console.print.assert_not_called()
def test_wait_for_mcp_browse(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/mcp"):
_run(ui.wait_for_mcp_browse([], set(), ""))
def test_force_quit_fires_callback(self):
called = []
ui, _ = _make_ui(on_force_quit=lambda: called.append("fq"))
ui.force_quit()
assert called == ["fq"]
def test_clear_chat(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/clear"):
ui.clear_chat()
def test_clear_chat_fires_callback(self):
called = []
ui, console = _make_ui(on_clear_chat=lambda: called.append("cls"))
ui.clear_chat()
assert called == ["cls"]
# Callback owns clearing — adapter should not also clear
console.clear.assert_not_called()
def test_request_quit(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/exit"):
ui.request_quit()
def test_clear_chat_default_falls_back_to_console_clear(self):
ui, console = _make_ui()
ui.clear_chat()
console.clear.assert_called_once()
def test_force_quit(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/exit"):
ui.force_quit()
def test_update_status_after_compact_fires_callback(self):
received = []
ui, _ = _make_ui(on_status_after_compact=received.append)
ui.update_status_after_compact(1234)
assert received == [1234]
def test_start_new_session(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/new"):
ui.start_new_session()
def test_update_status_after_compact_without_callback_is_noop(self):
ui, console = _make_ui()
# Does not raise; does not print.
ui.update_status_after_compact(500)
console.print.assert_not_called()
class TestWaitForThreadPick:
"""Phase B questionary picker — ported from the pre-migration
``_cmd_resume`` implementation in ``interactive.py``."""
def _fake_prompt(self, selected):
"""Build a stub matching ``questionary.select(...)`` return."""
from unittest.mock import AsyncMock
prompt = MagicMock()
# The adapter uses ``ask_async`` (questionary >= 2.0.1) so we
# must stub the async variant. Keep ``ask`` around in case
# other code paths still call it.
prompt.ask_async = AsyncMock(return_value=selected)
prompt.ask.return_value = selected
prompt.application.layout.find_all_windows.return_value = []
return prompt
def _threads(self):
return [
{
"thread_id": "abc123",
"preview": "hello",
"message_count": 3,
"model": "claude",
"updated_at": None,
"workspace_dir": "/w",
},
{
"thread_id": "def456",
"preview": "x" * 60, # triggers truncation path
"message_count": 0,
"model": None,
"updated_at": None,
"workspace_dir": "/w",
},
]
def test_returns_selected_thread_id(self, monkeypatch):
import EvoScientist.cli.rich_command_ui as mod
def test_handle_session_resume(self):
ui, _ = _make_ui()
with pytest.raises(NotImplementedError, match="/resume"):
_run(ui.handle_session_resume("tid"))
prompt = self._fake_prompt("abc123")
called = {}
def fake_select(title, choices, style):
called["title"] = title
called["choices"] = choices
return prompt
monkeypatch.setattr("questionary.select", fake_select)
result = _run(ui.wait_for_thread_pick(self._threads(), "abc123", "pick:"))
assert result == "abc123"
assert called["title"] == "pick:"
# _build_items prepends a workspace header — choices has headers +
# thread Choice entries.
assert len(called["choices"]) >= 2
# Table import removed; this test no longer depends on console output.
assert mod.RichCLICommandUI is not None # sanity
def test_cancel_returns_none(self, monkeypatch):
ui, _ = _make_ui()
prompt = self._fake_prompt(None)
monkeypatch.setattr("questionary.select", lambda *a, **k: prompt)
result = _run(ui.wait_for_thread_pick(self._threads(), "abc123", "pick:"))
assert result is None
def test_current_thread_marker_in_label(self, monkeypatch):
ui, _ = _make_ui()
prompt = self._fake_prompt(None)
captured_choices: list = []
def fake_select(title, choices, style):
captured_choices.extend(choices)
return prompt
monkeypatch.setattr("questionary.select", fake_select)
_run(ui.wait_for_thread_pick(self._threads(), "abc123", "pick:"))
# At least one Choice title contains "abc123 *" (current marker)
choice_titles = [getattr(c, "title", "") for c in captured_choices]
assert any("abc123 *" in t for t in choice_titles)
class TestCompactIndicator:
"""start/stop_compacting_indicator duck-typed by CompactCommand."""
def test_indicator_pair_wraps_console_status(self):
ui, console = _make_ui()
# Simulate a context manager returned by console.status()
status_cm = MagicMock()
console.status.return_value = status_cm
ui.start_compacting_indicator()
console.status.assert_called_once()
status_cm.__enter__.assert_called_once()
ui.stop_compacting_indicator()
status_cm.__exit__.assert_called_once_with(None, None, None)
def test_stop_without_start_is_noop(self):
ui, console = _make_ui()
# Should not raise even if start was never called.
ui.stop_compacting_indicator()
console.status.assert_not_called()
class TestPhaseBMigrated:
"""Session lifecycle callbacks (start/resume) filled in Phase B."""
def test_start_new_session_fires_callback(self):
called = []
ui, _ = _make_ui(on_start_new_session=lambda: called.append("new"))
ui.start_new_session()
assert called == ["new"]
def test_start_new_session_without_callback_is_noop(self):
ui, console = _make_ui()
ui.start_new_session()
console.print.assert_not_called()
def test_handle_session_resume_awaits_callback(self):
from unittest.mock import AsyncMock
cb = AsyncMock()
ui, _ = _make_ui(on_handle_session_resume=cb)
_run(ui.handle_session_resume("tid-x", "/workspace"))
cb.assert_awaited_once_with("tid-x", "/workspace")
def test_handle_session_resume_without_callback_is_noop(self):
ui, _ = _make_ui()
# Should not raise
_run(ui.handle_session_resume("tid-x"))
def test_handle_session_resume_workspace_defaults_none(self):
from unittest.mock import AsyncMock
cb = AsyncMock()
ui, _ = _make_ui(on_handle_session_resume=cb)
_run(ui.handle_session_resume("tid-x"))
cb.assert_awaited_once_with("tid-x", None)
class TestPhaseCMigrated:
"""Skill/MCP browse pickers delegate to questionary helpers via
``asyncio.to_thread`` since questionary blocks the event loop."""
def test_skill_browse_delegates_to_picker(self, monkeypatch):
from unittest.mock import MagicMock
picker = MagicMock(return_value=["skill-a", "skill-b"])
monkeypatch.setattr(
"EvoScientist.cli.skills_cmd._pick_skills_interactive",
picker,
)
ui, _ = _make_ui()
result = _run(ui.wait_for_skill_browse([{"name": "a"}], {"installed"}, "core"))
assert result == ["skill-a", "skill-b"]
picker.assert_called_once_with([{"name": "a"}], {"installed"}, "core")
def test_skill_browse_cancel_returns_none(self, monkeypatch):
from unittest.mock import MagicMock
monkeypatch.setattr(
"EvoScientist.cli.skills_cmd._pick_skills_interactive",
MagicMock(return_value=None),
)
ui, _ = _make_ui()
result = _run(ui.wait_for_skill_browse([], set(), ""))
assert result is None
def test_mcp_browse_delegates_to_picker(self, monkeypatch):
from unittest.mock import MagicMock
sentinel_entries = [MagicMock(name="entry1"), MagicMock(name="entry2")]
picker = MagicMock(return_value=sentinel_entries)
monkeypatch.setattr(
"EvoScientist.cli.mcp_install_cmd._browse_and_select",
picker,
)
ui, _ = _make_ui()
result = _run(ui.wait_for_mcp_browse([MagicMock()], {"configured"}, ""))
assert result is sentinel_entries
picker.assert_called_once()
def test_mcp_browse_cancel_returns_none(self, monkeypatch):
from unittest.mock import MagicMock
monkeypatch.setattr(
"EvoScientist.cli.mcp_install_cmd._browse_and_select",
MagicMock(return_value=None),
)
ui, _ = _make_ui()
result = _run(ui.wait_for_mcp_browse([], set(), ""))
assert result is None
tests/test_threads_command.py
+113
View File
@@ -0,0 +1,113 @@
"""Tests for the /threads command."""
from unittest.mock import AsyncMock, MagicMock, patch
from rich.table import Table
from tests.conftest import run_async as _run
def _ctx(**overrides):
from EvoScientist.commands.base import CommandContext
ui = MagicMock()
ui.supports_interactive = overrides.pop("supports_interactive", True)
return CommandContext(
agent=None,
thread_id=overrides.pop("thread_id", "tid-1"),
ui=ui,
workspace_dir=overrides.pop("workspace_dir", "/ws"),
), ui
class TestThreadsCommand:
def test_empty_list_prints_message(self):
from EvoScientist.commands.implementation.session import ThreadsCommand
ctx, ui = _ctx()
with patch(
"EvoScientist.sessions.list_threads",
new=AsyncMock(return_value=[]),
):
_run(ThreadsCommand().execute(ctx, []))
ui.append_system.assert_called_once()
assert "No saved sessions" in ui.append_system.call_args.args[0]
def test_renders_table_with_current_marker(self):
from EvoScientist.commands.implementation.session import ThreadsCommand
ctx, ui = _ctx(thread_id="current")
threads = [
{
"thread_id": "current",
"preview": "foo",
"message_count": 5,
"model": "claude",
"updated_at": None,
},
{
"thread_id": "other",
"preview": "bar",
"message_count": 2,
"model": None,
"updated_at": None,
},
]
with patch(
"EvoScientist.sessions.list_threads",
new=AsyncMock(return_value=threads),
):
_run(ThreadsCommand().execute(ctx, []))
ui.mount_renderable.assert_called_once()
table = ui.mount_renderable.call_args.args[0]
assert isinstance(table, Table)
# Footer hint (ported from the pre-migration inline /threads handler)
footer = ui.append_system.call_args.args[0]
assert "/resume" in footer
assert "/delete" in footer
assert "/new" in footer
def test_footer_hint_suppressed_in_channel_mode(self):
"""Channels don't get the footer — keeps outbound text short."""
from EvoScientist.commands.implementation.session import ThreadsCommand
ctx, ui = _ctx(supports_interactive=False)
threads = [
{
"thread_id": "t",
"preview": "p",
"message_count": 1,
"model": "m",
"updated_at": None,
}
]
with patch(
"EvoScientist.sessions.list_threads",
new=AsyncMock(return_value=threads),
):
_run(ThreadsCommand().execute(ctx, []))
ui.append_system.assert_not_called()
def test_channel_mode_drops_model_column(self):
"""Non-interactive (channel) UIs get a narrower table."""
from EvoScientist.commands.implementation.session import ThreadsCommand
ctx, ui = _ctx(supports_interactive=False)
threads = [
{
"thread_id": "t",
"preview": "p",
"message_count": 1,
"model": "m",
"updated_at": None,
}
]
with patch(
"EvoScientist.sessions.list_threads",
new=AsyncMock(return_value=threads),
):
_run(ThreadsCommand().execute(ctx, []))
# Channel mode: no Model column. 4 columns: ID, Preview, Msgs, Last Used.
table = ui.mount_renderable.call_args.args[0]
column_headers = [col.header for col in table.columns]
assert "Model" not in column_headers