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

feat: extract commands from TUI to commands module

Browse Source 
- adjust discord max message length to 2000 (the actual limit)
- /install-skill will fallback to searching EvoSkills repository if path not found to allow for installing skills from channels
- remove discord config tests (it asserted @dataclass behaviour)
This commit is contained in:
Jan Piotrowski
2026-03-18 12:04:43 +01:00
parent 1c6f4a2078
commit f9734c7e67
16 changed files with 1514 additions and 836 deletions
Download Patch File Download Diff File Expand all files Collapse all files
EvoScientist/channels/base.py
+66 -47
View File
@@ -29,17 +29,10 @@ _logger = logging.getLogger(__name__)
def chunk_text(text: str, limit: int) -> list[str]:
"""Split text into chunks that respect logical boundaries.
"""Split text into chunks that respect logical boundaries and code fences.
Splitting priority (highest to lowest):
1. Code block boundaries (``` fences)
2. Double newlines (paragraph breaks)
3. Single newlines
4. Spaces (word boundaries)
5. Hard cut (last resort)
Code blocks are never split mid-block when possible. If a single code
block exceeds the limit it is sent as its own chunk(s).
If a code block is split across chunks, each chunk is automatically
wrapped in its own fences (```...```) to maintain formatting.
Args:
text: The text to split.
@@ -55,49 +48,75 @@ def chunk_text(text: str, limit: int) -> list[str]:
chunks: list[str] = []
remaining = text
in_code_block = False
code_block_lang = ""
while remaining:
if len(remaining) <= limit:
chunks.append(remaining)
break
# Effective limit is reduced if we need to add fences
# We reserve ~20 chars for fences (```lang\n and \n```)
effective_limit = limit - (20 if in_code_block else 0)
if len(remaining) <= effective_limit:
segment = remaining
best = len(remaining)
else:
segment = remaining[:effective_limit]
best = -1
# Try to find a split point within the limit
segment = remaining[:limit]
# 1. Paragraph/Line/Word boundaries
if not in_code_block:
# Paragraph
pos = segment.rfind("\n\n")
if pos > 0:
best = pos
# Line
if best == -1:
pos = segment.rfind("\n")
if pos > 0:
best = pos
# Word
if best == -1:
pos = segment.rfind(" ")
if pos > 0:
best = pos
else:
# INSIDE code block: ONLY split at newlines to avoid breaking lines of code
pos = segment.rfind("\n")
if pos > 0:
best = pos
# 1. Prefer splitting at code block boundary (``` at line start)
best = -1
fence_pos = segment.rfind("\n```")
if fence_pos > 0:
line_end = segment.find("\n", fence_pos + 1)
if line_end == -1:
line_end = len(segment)
best = line_end
if best == -1:
best = effective_limit
# 2. Double newline (paragraph break)
if best == -1:
pos = segment.rfind("\n\n")
if pos > 0:
best = pos
chunk_raw = remaining[:best].rstrip()
# Track state transitions within this raw segment
starts_in_code = in_code_block
current_lang = code_block_lang
# We use a simple count of ``` to toggle state.
# Note: This handles both opening and closing fences.
fences = list(re.finditer(r"```(\w*)", chunk_raw))
for f in fences:
if not in_code_block:
in_code_block = True
code_block_lang = f.group(1) or ""
else:
in_code_block = False
code_block_lang = ""
ends_in_code = in_code_block
# 3. Single newline
if best == -1:
pos = segment.rfind("\n")
if pos > 0:
best = pos
# 4. Space (word boundary)
if best == -1:
pos = segment.rfind(" ")
if pos > 0:
best = pos
# 5. Hard cut
if best == -1:
best = limit
chunk = remaining[:best].rstrip()
if chunk:
chunks.append(chunk)
# Build the final chunk with necessary fences
prefix = f"```{current_lang}\n" if starts_in_code else ""
suffix = "\n```" if ends_in_code else ""
final_chunk = prefix + chunk_raw + suffix
if final_chunk.strip():
chunks.append(final_chunk)
remaining = remaining[best:].lstrip("\n")
return chunks
EvoScientist/channels/discord/channel.py
+1 -1
View File
@@ -16,7 +16,7 @@ logger = logging.getLogger(__name__)
@dataclass
class DiscordConfig(BaseChannelConfig):
bot_token: str = ""
text_chunk_limit: int = 4096
text_chunk_limit: int = 2000
class DiscordChannel(Channel):
EvoScientist/cli/tui_interactive.py
+136 -762
View File
@@ -10,11 +10,9 @@ import asyncio
import logging
import queue
import random
import shlex
from typing import Any, Callable
from rich.console import Group
from rich.table import Table
from rich.text import Text
import EvoScientist.cli.channel as _ch_mod
@@ -28,14 +26,11 @@ from .channel import (
_set_channel_response,
)
from ..sessions import (
_format_relative_time,
delete_thread,
find_similar_threads,
generate_thread_id,
get_checkpointer,
get_thread_messages,
get_thread_metadata,
list_threads,
thread_exists,
)
from ..config.settings import get_config_dir
@@ -43,28 +38,11 @@ from ..stream.events import stream_agent_events
from ..stream.state import StreamState, _INTERNAL_TOOLS
from .history_suggester import HistorySuggester
from ..commands import manager as cmd_manager, CommandContext
from ._constants import LOGO_LINES, LOGO_GRADIENT, WELCOME_SLOGANS, build_metadata
_channel_logger = logging.getLogger(__name__)
_TUI_SLASH_COMMANDS = [
("/current", "Show current session info"),
("/threads", "List recent sessions"),
("/resume", "Resume a previous session"),
("/delete", "Delete a saved session"),
("/new", "Start a new session"),
("/clear", "Clear chat history"),
("/skills", "List installed skills"),
("/install-skill", "Add a skill from path or GitHub"),
("/uninstall-skill", "Remove an installed skill"),
("/install-skills", "Browse and install skills (optional: /install-skills <tag>)"),
("/mcp", "Manage MCP servers"),
("/channel", "Configure messaging channels"),
("/compact", "Compact conversation to free context"),
("/help", "Show available commands"),
("/exit", "Quit EvoScientist"),
]
def _shorten_path(path: str) -> str:
"""Shorten absolute path to a cwd-relative form (consistent with Rich CLI)."""
@@ -220,6 +198,10 @@ def run_textual_interactive(
class EvoTextualInteractiveApp(App[None]): # type: ignore[type-arg]
"""Deep-Agents-style full-screen TUI with independent widget rendering."""
@property
def supports_interactive(self) -> bool:
return True
CSS = """
Screen {
layout: vertical;
@@ -326,6 +308,99 @@ def run_textual_interactive(
self._browser_future: asyncio.Future | None = None
self._history_suggester = HistorySuggester(get_config_dir() / "history")
# ── CommandUI implementation ─────────────────────────
def append_system(self, text: str, style: str = "dim") -> None:
self._append_system(text, style)
def mount_renderable(self, renderable: Any) -> None:
self._mount_renderable(renderable)
async def wait_for_thread_pick(
self, threads: list[dict], current_thread: str, title: str
) -> str | None:
from .widgets.thread_selector import ThreadPickerWidget
container = self.query_one("#chat", VerticalScroll)
picker = ThreadPickerWidget(
threads,
current_thread=current_thread,
title=title,
)
await container.mount(picker)
container.scroll_end(animate=False)
picker.focus()
return await self._wait_for_thread_pick(picker)
async def wait_for_skill_browse(
self, index: list[dict], installed_names: set[str], pre_filter_tag: str
) -> list[str] | None:
from .widgets.skill_browser import SkillBrowserWidget
container = self.query_one("#chat", VerticalScroll)
browser = SkillBrowserWidget(
index,
installed_names,
pre_filter_tag=pre_filter_tag,
)
await container.mount(browser)
container.scroll_end(animate=False)
browser.focus()
return await self._wait_for_skill_browse(browser)
def clear_chat(self) -> None:
container = self.query_one("#chat", VerticalScroll)
welcome = self.query_one("#welcome", Static)
for child in list(container.children):
if child is not welcome:
child.remove()
def request_quit(self) -> None:
self.action_request_quit()
def start_new_session(self) -> None:
# Clear all widgets except #welcome
self.clear_chat()
if not workspace_fixed:
self._workspace_dir = create_session_workspace(run_name)
self._conversation_tid = generate_thread_id()
self._agent = load_agent(
workspace_dir=self._workspace_dir,
checkpointer=self._checkpointer,
)
if _channels_is_running():
_ch_mod._cli_agent = self._agent
_ch_mod._cli_thread_id = self._conversation_tid
self._render_welcome()
self._render_status()
self.append_system(
f"New session: {self._conversation_tid}", style="green"
)
async def handle_session_resume(self, thread_id: str, workspace_dir: str | None = None) -> None:
if workspace_dir:
self._workspace_dir = workspace_dir
self._conversation_tid = thread_id
self._agent = load_agent(
workspace_dir=self._workspace_dir,
checkpointer=self._checkpointer,
)
if _channels_is_running():
_ch_mod._cli_agent = self._agent
_ch_mod._cli_thread_id = self._conversation_tid
self._render_welcome()
self._render_status()
self.append_system(f"Resumed session: {thread_id}", style="green")
await self._render_history(thread_id)
async def flush(self) -> None:
"""No-op for TUI, messages are already delivered incrementally."""
pass
# ── Layout ─────────────────────────────────────────────
def compose(self) -> ComposeResult:
@@ -406,7 +481,7 @@ def run_textual_interactive(
# ── Widget helpers ─────────────────────────────────────
def _append_system(self, text: str, *, style: str = "dim") -> None:
def _append_system(self, text: str, style: str = "dim") -> None:
"""Mount a SystemMessage widget into #chat."""
container = self.query_one("#chat", VerticalScroll)
container.mount(SystemMessage(text, msg_style=style))
@@ -1316,6 +1391,34 @@ def run_textual_interactive(
"""
return _ch_mod.channel_ask_user_prompt(ask_user_data, msg)
from ..commands.channel_ui import ChannelCommandUI
# Handle slash commands from channel
if msg.content.strip().startswith("/"):
ctx = CommandContext(
agent=self._agent,
thread_id=self._conversation_tid,
ui=ChannelCommandUI(
msg,
append_system_callback=self._append_system,
start_new_session_callback=self.start_new_session,
handle_session_resume_callback=self.handle_session_resume,
),
workspace_dir=self._workspace_dir,
checkpointer=self._checkpointer,
)
if await cmd_manager.execute(msg.content, ctx):
self._append_system(
f"[{msg.channel_type}: Executed command from {msg.sender}]",
style="dim",
)
_set_channel_response(msg.msg_id, f"Command executed: {msg.content}")
self._busy = False
self._render_status()
prompt_widget.disabled = False
prompt_widget.focus()
return
response = ""
try:
response = await self._stream_with_widgets(
@@ -1384,7 +1487,7 @@ def run_textual_interactive(
prefix = text.lower()
matches = [
(cmd, desc)
for cmd, desc in _TUI_SLASH_COMMANDS
for cmd, desc in cmd_manager.list_commands()
if cmd.startswith(prefix)
]
if len(matches) == 1 and matches[0][0] == prefix:
@@ -1572,170 +1675,22 @@ def run_textual_interactive(
# ── Slash commands ─────────────────────────────────────
async def _handle_command(self, command: str) -> None:
cmd, _, arg = command.strip().partition(" ")
cmd = cmd.lower()
arg = arg.strip()
# Echo the command so the user sees what they ran
self._append_system(command.strip(), style="cyan")
if cmd in ("/exit", "/quit", "/q"):
self.action_request_quit()
return
ctx = CommandContext(
agent=self._agent,
thread_id=self._conversation_tid,
ui=self,
workspace_dir=self._workspace_dir,
checkpointer=self._checkpointer,
)
if cmd == "/help":
help_text = Text("Available commands:\n", style="bold")
for hcmd, hdesc in _TUI_SLASH_COMMANDS:
help_text.append(f" {hcmd:<22}", style="cyan")
help_text.append(f"{hdesc}\n", style="dim")
self._mount_renderable(help_text)
return
if cmd == "/current":
from .. import paths
self._append_system(f"Thread: {self._conversation_tid}", style="dim")
if self._workspace_dir:
self._append_system(
f"Workspace: {_shorten_path(self._workspace_dir)}",
style="dim",
)
memory_path = _shorten_path(str(paths.MEMORY_DIR))
if memory_path:
self._append_system(f"Memory dir: {memory_path}", style="dim")
self._append_system("UI: tui", style="dim")
return
if cmd == "/new":
# Clear all widgets except #welcome
container = self.query_one("#chat", VerticalScroll)
welcome = self.query_one("#welcome", Static)
for child in list(container.children):
if child is not welcome:
await child.remove()
if not workspace_fixed:
self._workspace_dir = create_session_workspace(run_name)
self._conversation_tid = generate_thread_id()
self._agent = load_agent(
workspace_dir=self._workspace_dir,
checkpointer=self._checkpointer,
)
if _channels_is_running():
_ch_mod._cli_agent = self._agent
_ch_mod._cli_thread_id = self._conversation_tid
self._render_welcome()
self._render_status()
self._append_system(
f"New session: {self._conversation_tid}", style="green"
)
return
if cmd == "/clear":
container = self.query_one("#chat", VerticalScroll)
welcome = self.query_one("#welcome", Static)
for child in list(container.children):
if child is not welcome:
await child.remove()
return
if cmd == "/threads":
await self._cmd_threads()
return
if cmd == "/resume":
await self._cmd_resume(arg)
return
if cmd == "/delete":
await self._cmd_delete(arg)
return
if cmd == "/skills":
self._cmd_skills()
return
if cmd == "/install-skill":
self._cmd_install_skill(arg)
return
if cmd == "/uninstall-skill":
self._cmd_uninstall_skill(arg)
return
if cmd == "/install-skills":
await self._cmd_install_skills(arg)
return
if cmd == "/mcp":
self._cmd_mcp(arg)
return
if cmd == "/channel":
self._cmd_channel(arg)
return
if cmd == "/compact":
from .commands import compact_conversation, render_compact_result
self._append_system("Compacting conversation...")
result = await compact_conversation(
agent=self._agent,
thread_id=self._conversation_tid,
)
self._mount_renderable(render_compact_result(result))
if await cmd_manager.execute(command, ctx):
return
self._append_system(f"Unknown command: {command}", style="yellow")
async def _resolve_thread_id(self, prefix: str) -> str | None:
if await thread_exists(prefix):
return prefix
similar = await find_similar_threads(prefix)
if len(similar) == 1:
return similar[0]
if len(similar) > 1:
self._append_system(
f"Ambiguous thread ID '{prefix}'. Use a longer prefix.",
style="yellow",
)
for thread in similar:
self._append_system(f" - {thread}", style="dim")
return None
self._append_system(f"Thread '{prefix}' not found.", style="red")
return None
async def _cmd_threads(self) -> None:
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
self._append_system("No saved sessions.", style="yellow")
return
table = Table(title="Sessions", show_header=True, header_style="bold cyan")
table.add_column("ID", style="bold")
table.add_column("Preview", style="dim", max_width=50, no_wrap=True)
table.add_column("Messages", justify="right")
table.add_column("Model", style="dim")
table.add_column("Last Used", style="dim")
for thread in threads:
thread_id_value = thread["thread_id"]
marker = " *" if thread_id_value == self._conversation_tid else ""
table.add_row(
f"{thread_id_value}{marker}",
thread.get("preview", "") or "",
str(thread.get("message_count", 0)),
thread.get("model", "") or "",
_format_relative_time(thread.get("updated_at")),
)
self._mount_renderable(table)
async def _render_history(self, thread_id_value: str) -> None:
"""Render conversation history from a saved thread."""
messages = await get_thread_messages(thread_id_value)
@@ -1778,587 +1733,6 @@ def run_textual_interactive(
)
container.scroll_end(animate=False)
async def _cmd_resume(self, arg: str) -> None:
if not arg:
# Show inline thread picker
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
self._append_system("No sessions to resume.", style="yellow")
return
from .widgets.thread_selector import ThreadPickerWidget
container = self.query_one("#chat", VerticalScroll)
picker = ThreadPickerWidget(
threads,
current_thread=self._conversation_tid,
title=">>> Select session to resume <<<",
)
await container.mount(picker)
container.scroll_end(animate=False)
picker.focus()
selected = await self._wait_for_thread_pick(picker)
if selected is None:
return
arg = selected
resolved = await self._resolve_thread_id(arg)
if not resolved:
return
metadata = await get_thread_metadata(resolved)
restored_workspace = (metadata or {}).get("workspace_dir", "")
if restored_workspace:
self._workspace_dir = restored_workspace
self._conversation_tid = resolved
self._agent = load_agent(
workspace_dir=self._workspace_dir,
checkpointer=self._checkpointer,
)
if _channels_is_running():
_ch_mod._cli_agent = self._agent
_ch_mod._cli_thread_id = self._conversation_tid
self._render_welcome()
self._render_status()
self._append_system(f"Resumed session: {resolved}", style="green")
await self._render_history(resolved)
async def _cmd_delete(self, arg: str) -> None:
if not arg:
# Show inline thread picker for deletion
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
self._append_system("No sessions to delete.", style="yellow")
return
from .widgets.thread_selector import ThreadPickerWidget
container = self.query_one("#chat", VerticalScroll)
picker = ThreadPickerWidget(
threads,
current_thread=self._conversation_tid,
title=">>> Select session to delete <<<",
)
await container.mount(picker)
container.scroll_end(animate=False)
picker.focus()
selected = await self._wait_for_thread_pick(picker)
if selected is None:
return
arg = selected
resolved = await self._resolve_thread_id(arg)
if not resolved:
return
if resolved == self._conversation_tid:
self._append_system(
"Cannot delete the current session.",
style="yellow",
)
return
deleted = await delete_thread(resolved)
if deleted:
self._append_system(f"Deleted session {resolved}.", style="green")
else:
self._append_system(f"Session {resolved} not found.", style="red")
def _cmd_skills(self) -> None:
from ..tools.skills_manager import list_skills
from ..paths import USER_SKILLS_DIR
skills = list_skills(include_system=True)
if not skills:
self._append_system("No skills available.", style="dim")
self._append_system(
"Install with: /install-skill <path-or-url>", style="dim"
)
self._append_system(
f"Skills directory: {_shorten_path(str(USER_SKILLS_DIR))}",
style="dim",
)
return
user_skills = [s for s in skills if s.source == "user"]
system_skills = [s for s in skills if s.source == "system"]
if user_skills:
table = Table(
title=f"User Skills ({len(user_skills)})", show_header=True
)
table.add_column("Name", style="green")
table.add_column("Description", style="dim")
table.add_column("Tags", style="dim")
for s in user_skills:
tags = "\n".join(f"· {t}" for t in s.tags[:4]) if s.tags else ""
table.add_row(s.name, s.description, tags)
self._mount_renderable(table)
if system_skills:
table = Table(
title=f"Built-in Skills ({len(system_skills)})", show_header=True
)
table.add_column("Name", style="cyan")
table.add_column("Description", style="dim")
table.add_column("Tags", style="dim")
for s in system_skills:
tags = "\n".join(f"· {t}" for t in s.tags[:4]) if s.tags else ""
table.add_row(s.name, s.description, tags)
self._mount_renderable(table)
self._append_system(
f"User skills folder: {_shorten_path(str(USER_SKILLS_DIR))}",
style="dim",
)
def _cmd_install_skill(self, source: str) -> None:
from ..tools.skills_manager import install_skill
if not source:
self._append_system(
"Usage: /install-skill <path-or-url>", style="yellow"
)
self._append_system("Examples:", style="dim")
self._append_system(" /install-skill ./my-skill", style="dim")
self._append_system(
" /install-skill https://github.com/user/repo/tree/main/skill-name",
style="dim",
)
self._append_system(
" /install-skill user/repo@skill-name", style="dim"
)
return
self._append_system(f"Installing skill from: {source}", style="dim")
result = install_skill(source)
if result["success"]:
self._append_system(f"Installed: {result['name']}", style="green")
self._append_system(
f"Description: {result.get('description', '(none)')}",
style="dim",
)
self._append_system(
f"Path: {_shorten_path(result['path'])}", style="dim"
)
self._append_system("Reload with /new to apply.", style="dim")
else:
self._append_system(f"Failed: {result['error']}", style="red")
async def _cmd_install_skills(self, args: str) -> None:
from pathlib import Path as _Path
from ..paths import USER_SKILLS_DIR
from ..tools.skills_manager import fetch_remote_skill_index, install_skill
self._append_system("Fetching skill index...", style="dim")
try:
index = fetch_remote_skill_index()
except Exception as e:
self._append_system(f"Failed to fetch skill index: {e}", style="red")
self._append_system(
"Try: /install-skill EvoScientist/EvoSkills@skills", style="dim"
)
return
if not index:
self._append_system("No skills found.", style="yellow")
return
# Detect installed skills
skills_dir = _Path(USER_SKILLS_DIR)
installed_names: set[str] = set()
if skills_dir.exists():
installed_names = {e.name for e in skills_dir.iterdir() if e.is_dir()}
# Mount interactive browser widget
from .widgets.skill_browser import SkillBrowserWidget
container = self.query_one("#chat", VerticalScroll)
browser = SkillBrowserWidget(
index,
installed_names,
pre_filter_tag=args.strip(),
)
await container.mount(browser)
container.scroll_end(animate=False)
browser.focus()
# Wait for user interaction
selected_sources = await self._wait_for_skill_browse(browser)
if not selected_sources:
self._append_system("Browse cancelled.", style="dim")
return
# Install selected skills
installed_count = 0
for source in selected_sources:
result = install_skill(source)
if result.get("batch"):
for item in result.get("installed", []):
self._append_system(f"Installed: {item['name']}", style="green")
installed_count += 1
elif result.get("success"):
self._append_system(f"Installed: {result['name']}", style="green")
installed_count += 1
else:
self._append_system(
f"Failed: {result.get('error', 'unknown')}", style="red"
)
if installed_count:
self._append_system(
f"{installed_count} skill(s) installed. Reload with /new to apply.",
style="green",
)
def _cmd_uninstall_skill(self, name: str) -> None:
from ..tools.skills_manager import uninstall_skill
if not name:
self._append_system(
"Usage: /uninstall-skill <skill-name>", style="yellow"
)
self._append_system("Use /skills to see installed skills.", style="dim")
return
result = uninstall_skill(name)
if result["success"]:
self._append_system(f"Uninstalled: {name}", style="green")
self._append_system("Reload with /new to apply.", style="dim")
else:
self._append_system(f"Failed: {result['error']}", style="red")
def _cmd_mcp(self, args: str) -> None:
args = args.strip()
if not args or args == "list":
self._mcp_list()
return
parts = args.split(maxsplit=1)
subcmd = parts[0].lower()
subargs = parts[1] if len(parts) > 1 else ""
if subcmd == "config":
self._mcp_config(subargs.strip())
elif subcmd == "add":
self._mcp_add(subargs)
elif subcmd == "edit":
self._mcp_edit(subargs)
elif subcmd == "remove":
self._mcp_remove(subargs.strip())
else:
self._append_system("MCP commands:", style="bold")
self._append_system(
" /mcp List configured servers", style="dim"
)
self._append_system(
" /mcp list List configured servers", style="dim"
)
self._append_system(
" /mcp config Show detailed server config", style="dim"
)
self._append_system(" /mcp add ... Add a server", style="dim")
self._append_system(
" /mcp edit ... Edit an existing server", style="dim"
)
self._append_system(" /mcp remove ... Remove a server", style="dim")
def _mcp_list(self) -> None:
from ..mcp import load_mcp_config
from ..mcp.client import USER_MCP_CONFIG
config = load_mcp_config()
if not config:
self._append_system("No MCP servers configured.", style="dim")
self._append_system(
"Add one with: /mcp add <name> <command-or-url> [args...]",
style="dim",
)
return
table = Table(title="MCP Servers", show_header=True)
table.add_column("Server", style="cyan")
table.add_column("Transport", style="green")
table.add_column("Tools", style="yellow")
table.add_column("Expose To", style="magenta")
for name, server in config.items():
transport = server.get("transport", "?")
tools = server.get("tools")
tools_str = ", ".join(tools) if tools else "(all)"
expose_to = server.get("expose_to", ["main"])
if isinstance(expose_to, str):
expose_to = [expose_to]
expose_str = ", ".join(expose_to)
table.add_row(name, transport, tools_str, expose_str)
self._mount_renderable(table)
self._append_system(f"Config file: {USER_MCP_CONFIG}", style="dim")
def _mcp_config(self, name: str) -> None:
from ..mcp import load_mcp_config
from ..mcp.client import USER_MCP_CONFIG
config = load_mcp_config()
if not config:
self._append_system("No MCP servers configured.", style="dim")
return
if name and name not in config:
self._append_system(f"Server not found: {name}", style="red")
return
servers = {name: config[name]} if name else config
for srv_name, srv in servers.items():
table = Table(
title=f"MCP Server: {srv_name}",
show_header=True,
title_style="bold cyan",
)
table.add_column("Setting", style="cyan")
table.add_column("Value")
table.add_row("transport", str(srv.get("transport", "(not set)")))
if srv.get("command"):
table.add_row("command", str(srv["command"]))
if srv.get("args"):
table.add_row("args", " ".join(str(a) for a in srv["args"]))
if srv.get("url"):
table.add_row("url", str(srv["url"]))
if srv.get("headers"):
for k, v in srv["headers"].items():
table.add_row(f"header: {k}", str(v))
if srv.get("env"):
for k, v in srv["env"].items():
table.add_row(f"env: {k}", str(v))
tools = srv.get("tools")
table.add_row("tools", ", ".join(tools) if tools else "(all)")
expose_to = srv.get("expose_to", ["main"])
if isinstance(expose_to, str):
expose_to = [expose_to]
table.add_row("expose_to", ", ".join(expose_to))
self._mount_renderable(table)
self._append_system(f"Config file: {USER_MCP_CONFIG}", style="dim")
def _mcp_add(self, args_str: str) -> None:
from ..mcp import parse_mcp_add_args, add_mcp_server
if not args_str.strip():
self._append_system(
"Usage: /mcp add <name> <command-or-url> [args...]", style="yellow"
)
return
try:
tokens = shlex.split(args_str)
kwargs = parse_mcp_add_args(tokens)
entry = add_mcp_server(**kwargs)
self._append_system(
f"Added MCP server: {kwargs['name']} ({entry['transport']})",
style="green",
)
self._append_system("Reload with /new to apply.", style="dim")
except ValueError as exc:
self._append_system(f"{exc}", style="red")
def _mcp_edit(self, args_str: str) -> None:
from ..mcp import parse_mcp_edit_args, edit_mcp_server
if not args_str.strip():
self._append_system(
"Usage: /mcp edit <name> --<field> <value> ...",
style="yellow",
)
return
try:
tokens = shlex.split(args_str)
name, fields = parse_mcp_edit_args(tokens)
if not fields:
self._append_system(
"No fields to edit. Use --transport, --command, --url, --tools, --expose-to, etc.",
style="red",
)
return
edit_mcp_server(name, **fields)
self._append_system(f"Updated MCP server: {name}", style="green")
for k, v in fields.items():
self._append_system(f" {k}: {v}", style="dim")
self._append_system("Reload with /new to apply.", style="dim")
except (KeyError, ValueError) as exc:
self._append_system(f"{exc}", style="red")
def _mcp_remove(self, name: str) -> None:
from ..mcp import remove_mcp_server
if not name:
self._append_system("Usage: /mcp remove <name>", style="yellow")
return
if remove_mcp_server(name):
self._append_system(f"Removed MCP server: {name}", style="green")
self._append_system("Reload with /new to apply.", style="dim")
else:
self._append_system(f"Server not found: {name}", style="red")
def _cmd_channel(self, args: str) -> None:
"""Handle /channel command — start, stop, or show status."""
from ..config import load_config
from .channel import (
_add_channel_to_running_bus,
_start_channels_bus_mode,
_channels_running_list,
)
args = args.strip().lower() if args else ""
if args == "status" or (not args and _channels_is_running()):
running = _channels_running_list()
if running and _ch_mod._manager:
detailed = _ch_mod._manager.get_detailed_status()
table = Table(
title="Channel Status",
show_header=True,
expand=False,
)
table.add_column("Channel", style="cyan")
table.add_column("Status")
table.add_column("Uptime", style="dim")
table.add_column("Rx", justify="right")
table.add_column("Tx", justify="right")
for ch_name in running:
info = detailed.get(ch_name, {})
secs = info.get("uptime_seconds", 0)
mins, s = divmod(int(secs), 60)
hours, mins = divmod(mins, 60)
uptime = f"{hours}h{mins:02d}m" if hours else f"{mins}m{s:02d}s"
rx = str(info.get("received", 0))
tx = str(info.get("sent", 0))
table.add_row(
ch_name,
"[green]running[/green]",
uptime,
rx,
tx,
)
self._mount_renderable(table)
else:
self._append_system("No channel running", style="dim")
return
if args.startswith("stop"):
stop_type = args[len("stop") :].strip() or None
if not _channels_is_running():
self._append_system("No channel running", style="dim")
return
if stop_type:
if not _channels_is_running(stop_type):
self._append_system(
f"{stop_type} is not running",
style="dim",
)
return
_channels_stop(stop_type)
if stop_type in self._started_channel_types:
self._started_channel_types.remove(stop_type)
self._append_system(f"{stop_type} stopped", style="dim")
else:
running = _channels_running_list()
_channels_stop()
self._started_channel_types.clear()
self._append_system(
f"{', '.join(running)} stopped",
style="dim",
)
self._render_welcome()
return
# Start channel(s)
app_config = load_config()
channel_type = (
args if args else (app_config.channel_enabled if app_config else "")
)
if not channel_type:
self._append_system("No channel configured.", style="yellow")
self._append_system(
"Run EvoSci onboard or specify: /channel telegram",
style="dim",
)
return
requested = [t.strip() for t in channel_type.split(",") if t.strip()]
if _channels_is_running():
running = _channels_running_list()
results: list[tuple[str, bool, str]] = []
for ct in requested:
if ct in running:
results.append((ct, True, "already running"))
else:
try:
_add_channel_to_running_bus(
ct,
app_config,
send_thinking=self._channel_send_thinking,
)
results.append((ct, True, "connected (bus)"))
except Exception as e:
results.append((ct, False, str(e)))
else:
_ch_mod._cli_agent = self._agent
_ch_mod._cli_thread_id = self._conversation_tid
original = app_config.channel_enabled
app_config.channel_enabled = channel_type
try:
_start_channels_bus_mode(
app_config,
self._agent,
self._conversation_tid,
send_thinking=self._channel_send_thinking,
)
results = [(ct, True, "connected (bus)") for ct in requested]
except Exception as e:
results = [(ct, False, str(e)) for ct in requested]
finally:
app_config.channel_enabled = original
for ct, ok, _ in results:
if ok and ct not in self._started_channel_types:
self._started_channel_types.append(ct)
self._render_channel_results(results)
self._render_welcome()
def _render_channel_results(
self,
results: list[tuple[str, bool, str]],
) -> None:
for name, ok, detail in results:
if ok:
self._append_system(
f"\u25cf {name} {detail}",
style="green",
)
else:
self._append_system(
f"\u2717 {name} {detail}",
style="yellow",
)
# ── Quit handling ──────────────────────────────────────
def action_request_quit(self) -> None:
EvoScientist/commands/__init__.py
+17
View File
@@ -0,0 +1,17 @@
from __future__ import annotations
from . import implementation
from .base import Argument, Command, CommandContext, CommandUI
from .channel_ui import ChannelCommandUI
from .manager import CommandManager, manager
__all__ = [
"Argument",
"Command",
"CommandContext",
"CommandUI",
"manager",
"CommandManager",
"ChannelCommandUI",
"implementation",
]
EvoScientist/commands/base.py
+68
View File
@@ -0,0 +1,68 @@
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Protocol, Type, runtime_checkable
@dataclass
class Argument:
"""Definition of a command argument."""
name: str
type: Type
description: str
required: bool = True
@runtime_checkable
class CommandUI(Protocol):
"""Protocol for UI operations that commands can perform."""
@property
def supports_interactive(self) -> bool: ...
def append_system(self, text: str, style: str = "dim") -> None: ...
def mount_renderable(self, renderable: Any) -> None: ...
# Optional interactive operations
async def wait_for_thread_pick(
self, threads: list[dict], current_thread: str, title: str
) -> str | None: ...
async def wait_for_skill_browse(
self, index: list[dict], installed_names: set[str], pre_filter_tag: str
) -> list[str] | None: ...
def clear_chat(self) -> None: ...
def request_quit(self) -> None: ...
def start_new_session(self) -> None: ...
async def handle_session_resume(
self, thread_id: str, workspace_dir: str | None = None
) -> None: ...
async def flush(self) -> None: ...
@dataclass
class CommandContext:
"""Context passed to commands during execution."""
agent: Any
thread_id: str
ui: CommandUI
workspace_dir: str | None = None
checkpointer: Any = None
config: Any = None
# Add other fields as needed (e.g., current model, provider)
class Command(ABC):
"""Base class for all EvoScientist slash commands."""
name: str
alias: list[str] = []
description: str
arguments: list[Argument] = []
@abstractmethod
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
"""Execute the command with given context and arguments."""
pass
EvoScientist/commands/channel_ui.py
+141
View File
@@ -0,0 +1,141 @@
from __future__ import annotations
import asyncio
from typing import Any
from .base import CommandUI
class ChannelCommandUI(CommandUI):
"""CommandUI implementation for messaging channels with output buffering."""
@property
def supports_interactive(self) -> bool:
return False
def __init__(
self,
channel_msg: Any,
append_system_callback: Any = None,
start_new_session_callback: Any = None,
handle_session_resume_callback: Any = None,
):
self.msg = channel_msg
self.append_system_callback = append_system_callback
self.start_new_session_callback = start_new_session_callback
self.handle_session_resume_callback = handle_session_resume_callback
self._system_buffer: list[str] = []
def append_system(self, text: str, style: str = "dim") -> None:
if self.append_system_callback:
self.append_system_callback(text, style)
# Buffer the text for grouped delivery to the channel
# We ignore style for grouping but keep it for individual lines if needed
self._system_buffer.append(text)
async def flush(self) -> None:
"""Send all buffered system messages as a single grouped message."""
if not self._system_buffer:
return
grouped_text = "\n".join(self._system_buffer)
self._system_buffer = []
from ..channels.base import OutboundMessage
from ..cli.channel import _bus_loop
loop = _bus_loop
if not loop:
return
outbound = OutboundMessage(
channel=self.msg.channel_type,
chat_id=self.msg.chat_id,
content=grouped_text,
reply_to=self.msg.message_id,
metadata=self.msg.metadata,
)
if self.msg.bus_ref:
coro = self.msg.bus_ref.publish_outbound(outbound)
else:
coro = self.msg.channel_ref.send(outbound)
asyncio.run_coroutine_threadsafe(coro, loop)
def mount_renderable(self, renderable: Any) -> None:
# Convert Rich renderable to text/markdown for channel
from io import StringIO
from rich.console import Console
# Increase width to 120 to avoid wrapping tables like /threads
with StringIO() as f:
console = Console(
file=f, force_terminal=False, width=120, color_system=None
)
console.print(renderable)
text = f.getvalue()
from ..channels.base import OutboundMessage
from ..cli.channel import _bus_loop
loop = _bus_loop
if not loop:
return
# Flush any pending system messages first to preserve order
if self._system_buffer:
asyncio.run_coroutine_threadsafe(self.flush(), loop)
outbound = OutboundMessage(
channel=self.msg.channel_type,
chat_id=self.msg.chat_id,
content=f"```\n{text}\n```",
reply_to=self.msg.message_id,
metadata=self.msg.metadata,
)
if self.msg.bus_ref:
coro = self.msg.bus_ref.publish_outbound(outbound)
else:
coro = self.msg.channel_ref.send(outbound)
asyncio.run_coroutine_threadsafe(coro, loop)
async def wait_for_thread_pick(
self, threads: list[dict], current_thread: str, title: str
) -> str | None:
self.append_system(f"{title}\nUse /resume <id> to continue.")
await self.flush()
return None
async def wait_for_skill_browse(
self, index: list[dict], installed_names: set[str], pre_filter_tag: str
) -> list[str] | None:
self.append_system(
"Interactive skill browsing not supported in channels. Use /install-skill <name> instead."
)
await self.flush()
return None
def clear_chat(self) -> None:
self.append_system("Clear chat not supported in channels.")
def request_quit(self) -> None:
self.append_system("Quit command ignored in channel.")
def start_new_session(self) -> None:
if self.start_new_session_callback:
self.start_new_session_callback()
else:
self.append_system(
"New session requested. Please restart the channel link or use /new if supported."
)
async def handle_session_resume(
self, thread_id: str, workspace_dir: str | None = None
) -> None:
if self.handle_session_resume_callback:
await self.handle_session_resume_callback(thread_id, workspace_dir)
EvoScientist/commands/implementation/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from __future__ import annotations
from . import channel, general, mcp, session, skills
__all__ = ["general", "session", "skills", "mcp", "channel"]
EvoScientist/commands/implementation/channel.py
+165
View File
@@ -0,0 +1,165 @@
from __future__ import annotations
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from ..base import Command, CommandContext
from ..manager import manager
class ChannelCommand(Command):
"""Configure messaging channels."""
name = "/channel"
description = "Configure messaging channels"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
import EvoScientist.cli.channel as _ch_mod
from ...cli.channel import (
_channels_is_running,
_channels_running_list,
_channels_stop,
_start_channels_bus_mode,
)
from ...config import load_config
subcmd = args[0].lower() if args else ""
if subcmd == "status" or (not subcmd and _channels_is_running()):
running = _channels_running_list()
if running and _ch_mod._manager:
detailed = _ch_mod._manager.get_detailed_status()
table = Table(
title="Channel Status",
show_header=True,
header_style="bold cyan",
)
table.add_column("Channel")
table.add_column("Status")
table.add_column("Details", style="dim")
for name, info in detailed.items():
ok = info.get("running", False)
detail = (
f"sent: {info.get('sent', 0)}, recv: {info.get('received', 0)}"
)
status = (
"[green]\u25cf online[/green]"
if ok
else "[red]\u25cb offline[/red]"
)
table.add_row(name, status, detail)
ctx.ui.mount_renderable(table)
else:
ctx.ui.append_system(
"No messaging channels are currently active.", style="yellow"
)
return
if subcmd == "stop":
target = args[1] if len(args) > 1 else ""
if not _channels_is_running():
ctx.ui.append_system("No channels are running.", style="dim")
else:
if target:
_channels_stop(target)
ctx.ui.append_system(f"Channel '{target}' stopped.", style="green")
else:
_channels_stop()
ctx.ui.append_system("All channels stopped.", style="green")
return
config = load_config()
send_thinking = config.channel_send_thinking
# If args specifies channels, use them, otherwise use config
# Handle both space-separated and comma-separated inputs
requested = []
for a in args:
requested.extend([t.strip() for t in a.split(",") if t.strip()])
if _channels_is_running():
if not requested:
ctx.ui.append_system(
"Channels already running. Specify type to add: /channel <type>",
style="yellow",
)
return
ctx.ui.append_system(f"Adding channel(s): {', '.join(requested)}...")
from ...cli.channel import _add_channel_to_running_bus
try:
for ct in requested:
_add_channel_to_running_bus(ct, config, send_thinking=send_thinking)
ctx.ui.append_system("Channel(s) added successfully.", style="green")
except Exception as e:
ctx.ui.append_system(f"Failed to add channels: {e}", style="red")
return
ctx.ui.append_system("Starting channels...")
original = config.channel_enabled
if requested:
config.channel_enabled = ",".join(requested)
if not config.channel_enabled:
ctx.ui.append_system(
"No channels enabled in config. Use /channel <type> to start one.",
style="yellow",
)
ctx.ui.append_system(
"Types: telegram, discord, slack, feishu, dingtalk, wechat, email, imessage",
style="dim",
)
return
try:
# We need to make sure we don't block the UI.
# _start_channels_bus_mode might start threads/loops.
_start_channels_bus_mode(
config,
ctx.agent,
ctx.thread_id,
send_thinking=send_thinking,
)
# Show status panel
if _ch_mod._manager:
detailed = _ch_mod._manager.get_detailed_status()
all_ok = all(info.get("running", False) for info in detailed.values())
lines = []
for name, info in detailed.items():
ok = info.get("running", False)
line = Text()
if ok:
line.append("\u25cf ", style="green")
line.append(name, style="bold")
else:
line.append("\u25cb ", style="red")
line.append(name, style="bold red")
detail = (
f"sent: {info.get('sent', 0)}, recv: {info.get('received', 0)}"
)
line.append(f" {detail}", style="dim")
lines.append(line)
body = Text("\n").join(lines)
border = "green" if all_ok else "yellow"
ctx.ui.mount_renderable(
Panel(
body,
title="[bold]Channels[/bold]",
border_style=border,
expand=False,
)
)
except Exception as e:
ctx.ui.append_system(f"Failed to start channels: {e}", style="red")
finally:
config.channel_enabled = original
# Register Channel command
manager.register(ChannelCommand())
EvoScientist/commands/implementation/general.py
+67
View File
@@ -0,0 +1,67 @@
from __future__ import annotations
from rich.text import Text
from ..base import Command, CommandContext
from ..manager import manager
class HelpCommand(Command):
"""Show available commands."""
name = "/help"
description = "Show available commands"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
help_text = Text("Available commands:\n", style="bold")
for cmd in sorted(manager.get_all_commands(), key=lambda c: c.name):
# Build usage hint from arguments
usage_parts = [cmd.name]
for arg in cmd.arguments:
if arg.required:
usage_parts.append(f"<{arg.name}>")
else:
usage_parts.append(f"[{arg.name}]")
usage = " ".join(usage_parts)
help_text.append(f" {usage:<30}", style="cyan")
desc = cmd.description
if cmd.alias:
desc += f" (aliases: {', '.join(cmd.alias)})"
help_text.append(f"{desc}\n", style="dim")
ctx.ui.mount_renderable(help_text)
class CurrentCommand(Command):
"""Show current session info."""
name = "/current"
description = "Show current session info"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ... import paths
ctx.ui.append_system(f"Thread: {ctx.thread_id}", style="dim")
if ctx.workspace_dir:
from ...cli.agent import _shorten_path
ctx.ui.append_system(
f"Workspace: {_shorten_path(ctx.workspace_dir)}",
style="dim",
)
memory_path = paths.MEMORY_DIR
if memory_path:
from ...cli.agent import _shorten_path
ctx.ui.append_system(
f"Memory dir: {_shorten_path(str(memory_path))}", style="dim"
)
# How to determine UI type here?
# Maybe ctx.ui has a name or we pass it in ctx.
# For now, let's keep it simple.
ctx.ui.append_system("UI: auto", style="dim")
# Register commands
manager.register(HelpCommand())
manager.register(CurrentCommand())
EvoScientist/commands/implementation/mcp.py
+181
View File
@@ -0,0 +1,181 @@
from __future__ import annotations
from rich.table import Table
from ..base import Command, CommandContext
from ..manager import manager
class MCPCommand(Command):
"""Manage MCP servers."""
name = "/mcp"
description = "Manage MCP servers"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
if not args or args[0] == "list":
await self._mcp_list(ctx)
return
subcmd = args[0].lower()
subargs = args[1:]
if subcmd == "config":
await self._mcp_config(ctx, subargs[0] if subargs else "")
elif subcmd == "add":
await self._mcp_add(ctx, subargs)
elif subcmd == "edit":
await self._mcp_edit(ctx, subargs)
elif subcmd == "remove":
await self._mcp_remove(ctx, subargs[0] if subargs else "")
else:
ctx.ui.append_system("MCP commands:", style="bold")
ctx.ui.append_system(
" /mcp List configured servers", style="dim"
)
ctx.ui.append_system(
" /mcp list List configured servers", style="dim"
)
ctx.ui.append_system(
" /mcp config Show detailed server config", style="dim"
)
ctx.ui.append_system(" /mcp add ... Add a server", style="dim")
ctx.ui.append_system(
" /mcp edit ... Edit an existing server", style="dim"
)
ctx.ui.append_system(" /mcp remove ... Remove a server", style="dim")
async def _mcp_list(self, ctx: CommandContext) -> None:
from ...mcp import load_mcp_config
from ...mcp.client import USER_MCP_CONFIG
config = load_mcp_config()
if not config:
ctx.ui.append_system("No MCP servers configured.", style="dim")
ctx.ui.append_system(
"Add one with: /mcp add <name> <command-or-url> [args...]",
style="dim",
)
return
table = Table(title="MCP Servers", show_header=True)
table.add_column("Server", style="cyan")
table.add_column("Transport", style="green")
table.add_column("Tools", style="yellow")
table.add_column("Expose To", style="magenta")
for name, server in config.items():
transport = server.get("transport", "?")
tools = server.get("tools")
tools_str = ", ".join(tools) if tools else "(all)"
expose_to = server.get("expose_to", ["main"])
if isinstance(expose_to, str):
expose_to = [expose_to]
expose_str = ", ".join(expose_to)
table.add_row(name, transport, tools_str, expose_str)
ctx.ui.mount_renderable(table)
ctx.ui.append_system(f"Config file: {USER_MCP_CONFIG}", style="dim")
async def _mcp_config(self, ctx: CommandContext, name: str) -> None:
from ...mcp import load_mcp_config
from ...mcp.client import USER_MCP_CONFIG
config = load_mcp_config()
if not config:
ctx.ui.append_system("No MCP servers configured.", style="dim")
return
if name and name not in config:
ctx.ui.append_system(f"Server not found: {name}", style="red")
return
servers = {name: config[name]} if name else config
for srv_name, srv in servers.items():
table = Table(
title=f"MCP Server: {srv_name}",
show_header=True,
title_style="bold cyan",
)
table.add_column("Setting", style="cyan")
table.add_column("Value")
table.add_row("transport", str(srv.get("transport", "(not set)")))
if srv.get("command"):
table.add_row("command", str(srv["command"]))
if srv.get("args"):
table.add_row("args", " ".join(str(a) for a in srv["args"]))
if srv.get("url"):
table.add_row("url", str(srv["url"]))
if srv.get("headers"):
headers_str = ", ".join(f"{k}: {v}" for k, v in srv["headers"].items())
table.add_row("headers", headers_str)
if srv.get("env"):
env_str = ", ".join(f"{k}={v}" for k, v in srv["env"].items())
table.add_row("env", env_str)
tools = srv.get("tools")
table.add_row("tools", ", ".join(tools) if tools else "[dim](all)[/dim]")
expose_to = srv.get("expose_to", ["main"])
if isinstance(expose_to, str):
expose_to = [expose_to]
table.add_row("expose_to", ", ".join(expose_to))
ctx.ui.mount_renderable(table)
ctx.ui.append_system(f"Config file: {USER_MCP_CONFIG}", style="dim")
async def _mcp_add(self, ctx: CommandContext, tokens: list[str]) -> None:
from ...mcp import add_mcp_server, parse_mcp_add_args
if not tokens:
ctx.ui.append_system(
"Usage: /mcp add <name> <command-or-url> [args...]", style="yellow"
)
return
try:
kwargs = parse_mcp_add_args(tokens)
entry = add_mcp_server(**kwargs)
ctx.ui.append_system(
f"Added MCP server: {kwargs['name']} ({entry['transport']})",
style="green",
)
ctx.ui.append_system("Reload with /new to apply.", style="dim")
except Exception as exc:
ctx.ui.append_system(f"Error: {exc}", style="red")
async def _mcp_edit(self, ctx: CommandContext, tokens: list[str]) -> None:
from ...mcp import edit_mcp_server, parse_mcp_edit_args
if not tokens:
ctx.ui.append_system(
"Usage: /mcp edit <name> --<field> <value> ...", style="yellow"
)
return
try:
name, fields = parse_mcp_edit_args(tokens)
edit_mcp_server(name, **fields)
ctx.ui.append_system(f"Updated MCP server: {name}", style="green")
ctx.ui.append_system("Reload with /new to apply.", style="dim")
except Exception as exc:
ctx.ui.append_system(f"Error: {exc}", style="red")
async def _mcp_remove(self, ctx: CommandContext, name: str) -> None:
from ...mcp import remove_mcp_server
if not name:
ctx.ui.append_system("Usage: /mcp remove <name>", style="yellow")
return
if remove_mcp_server(name):
ctx.ui.append_system(f"Removed MCP server: {name}", style="green")
ctx.ui.append_system("Reload with /new to apply.", style="dim")
else:
ctx.ui.append_system(f"Server not found: {name}", style="red")
# Register MCP command
manager.register(MCPCommand())
EvoScientist/commands/implementation/session.py
+272
View File
@@ -0,0 +1,272 @@
from __future__ import annotations
from rich.table import Table
from ..base import Argument, Command, CommandContext
from ..manager import manager
class CompactCommand(Command):
"""Compact conversation to free context."""
name = "/compact"
description = "Compact conversation to free context"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ...cli.commands import compact_conversation, render_compact_result
ctx.ui.append_system("Compacting conversation...")
result = await compact_conversation(
agent=ctx.agent,
thread_id=ctx.thread_id,
)
ctx.ui.mount_renderable(render_compact_result(result))
class ThreadsCommand(Command):
"""List recent sessions."""
name = "/threads"
description = "List recent sessions"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ...sessions import _format_relative_time, list_threads
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
ctx.ui.append_system("No saved sessions.", style="yellow")
return
# Use protocol property to adapt output for non-interactive UIs (channels)
is_channel = not ctx.ui.supports_interactive
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=40 if is_channel else 50, no_wrap=True
)
table.add_column("Msgs" if is_channel else "Messages", justify="right")
if not is_channel:
table.add_column("Model", style="dim")
table.add_column("Last Used", style="dim")
for thread in threads:
thread_id_value = thread["thread_id"]
marker = " *" if thread_id_value == ctx.thread_id else ""
row = [
f"{thread_id_value}{marker}",
thread.get("preview", "") or "",
str(thread.get("message_count", 0)),
]
if not is_channel:
row.append(thread.get("model", "") or "")
row.append(_format_relative_time(thread.get("updated_at")))
table.add_row(*row)
ctx.ui.mount_renderable(table)
class ResumeCommand(Command):
"""Resume a previous session."""
name = "/resume"
description = "Resume a previous session"
arguments = [
Argument(
name="thread_id",
type=str,
description="Thread ID or prefix to resume",
required=False,
)
]
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ...sessions import (
get_thread_metadata,
list_threads,
)
arg = args[0] if args else ""
if not arg:
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
ctx.ui.append_system("No sessions to resume.", style="yellow")
return
# Interactive pick
selected = await ctx.ui.wait_for_thread_pick(
threads,
current_thread=ctx.thread_id,
title=">>> Select session to resume <<<",
)
if selected is None:
return
arg = selected
# Resolve thread_id
resolved = await self._resolve_thread_id(arg, ctx)
if not resolved:
return
metadata = await get_thread_metadata(resolved)
restored_workspace = (metadata or {}).get("workspace_dir", "")
if restored_workspace:
ctx.workspace_dir = restored_workspace
ctx.thread_id = resolved
# Signal session change to UI
if hasattr(ctx.ui, "handle_session_resume"):
await ctx.ui.handle_session_resume(resolved, restored_workspace)
async def _resolve_thread_id(self, prefix: str, ctx: CommandContext) -> str | None:
from ...sessions import find_similar_threads, thread_exists
if await thread_exists(prefix):
return prefix
similar = await find_similar_threads(prefix)
if len(similar) == 1:
return similar[0]
if len(similar) > 1:
ctx.ui.append_system(
f"Ambiguous thread ID '{prefix}'. Use a longer prefix.",
style="yellow",
)
for thread in similar:
ctx.ui.append_system(f" - {thread}", style="dim")
return None
ctx.ui.append_system(f"Thread '{prefix}' not found.", style="red")
return None
class NewCommand(Command):
"""Start a new session."""
name = "/new"
description = "Start a new session"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
ctx.ui.start_new_session()
class ClearCommand(Command):
"""Clear chat history."""
name = "/clear"
description = "Clear chat history"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
ctx.ui.clear_chat()
class DeleteCommand(Command):
"""Delete a saved session."""
name = "/delete"
description = "Delete a saved session"
arguments = [
Argument(
name="thread_id",
type=str,
description="Thread ID or prefix to delete",
required=False,
)
]
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ...sessions import (
delete_thread,
find_similar_threads,
list_threads,
thread_exists,
)
arg = args[0] if args else ""
if not arg:
threads = await list_threads(
limit=0,
include_message_count=True,
include_preview=True,
)
if not threads:
ctx.ui.append_system("No sessions to delete.", style="yellow")
return
# Interactive pick
selected = await ctx.ui.wait_for_thread_pick(
threads,
current_thread=ctx.thread_id,
title=">>> Select session to delete <<<",
)
if selected is None:
return
arg = selected
# Resolve thread_id
resolved = None
if await thread_exists(arg):
resolved = arg
else:
similar = await find_similar_threads(arg)
if len(similar) == 1:
resolved = similar[0]
elif len(similar) > 1:
ctx.ui.append_system(
f"Ambiguous thread ID '{arg}'. Use a longer prefix.",
style="yellow",
)
for thread in similar:
ctx.ui.append_system(f" - {thread}", style="dim")
return
if not resolved:
ctx.ui.append_system(f"Session '{arg}' not found.", style="red")
return
if resolved == ctx.thread_id:
ctx.ui.append_system(
"Cannot delete the current session.",
style="yellow",
)
return
deleted = await delete_thread(resolved)
if deleted:
ctx.ui.append_system(f"Deleted session {resolved}.", style="green")
else:
ctx.ui.append_system(f"Session {resolved} not found.", style="red")
class ExitCommand(Command):
"""Quit EvoScientist."""
name = "/exit"
alias = ["/quit", "/q"]
description = "Quit EvoScientist"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
ctx.ui.request_quit()
# Register session commands
manager.register(CompactCommand())
manager.register(ThreadsCommand())
manager.register(ResumeCommand())
manager.register(NewCommand())
manager.register(ClearCommand())
manager.register(DeleteCommand())
manager.register(ExitCommand())
EvoScientist/commands/implementation/skills.py
+246
View File
@@ -0,0 +1,246 @@
from __future__ import annotations
from rich.table import Table
from ..base import Argument, Command, CommandContext
from ..manager import manager
class SkillsCommand(Command):
"""List installed skills."""
name = "/skills"
description = "List installed skills"
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ...cli.agent import _shorten_path
from ...paths import USER_SKILLS_DIR
from ...tools.skills_manager import list_skills
skills = list_skills(include_system=True)
if not skills:
ctx.ui.append_system("No skills available.", style="dim")
ctx.ui.append_system(
"Install with: /install-skill <path-or-url>", style="dim"
)
ctx.ui.append_system(
f"Skills directory: {_shorten_path(str(USER_SKILLS_DIR))}",
style="dim",
)
return
user_skills = [s for s in skills if s.source == "user"]
system_skills = [s for s in skills if s.source == "system"]
if user_skills:
table = Table(title=f"User Skills ({len(user_skills)})", show_header=True)
table.add_column("Name", style="green")
table.add_column("Description", style="dim")
table.add_column("Tags", style="dim")
for s in user_skills:
tags = "\n".join(f"· {t}" for t in s.tags[:4]) if s.tags else ""
table.add_row(s.name, s.description, tags)
ctx.ui.mount_renderable(table)
if system_skills:
table = Table(
title=f"Built-in Skills ({len(system_skills)})", show_header=True
)
table.add_column("Name", style="cyan")
table.add_column("Description", style="dim")
table.add_column("Tags", style="dim")
for s in system_skills:
tags = "\n".join(f"· {t}" for t in s.tags[:4]) if s.tags else ""
table.add_row(s.name, s.description, tags)
ctx.ui.mount_renderable(table)
ctx.ui.append_system(
f"User skills folder: {_shorten_path(str(USER_SKILLS_DIR))}",
style="dim",
)
class InstallSkillCommand(Command):
"""Add a skill from path or GitHub."""
name = "/install-skill"
description = "Add a skill from path or GitHub"
arguments = [
Argument(
name="source",
type=str,
description="Path or GitHub URL of the skill",
required=True,
)
]
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ...cli.agent import _shorten_path
from ...tools.skills_manager import install_skill
source = args[0] if args else ""
if not source:
ctx.ui.append_system("Usage: /install-skill <path-or-url>", style="yellow")
ctx.ui.append_system("Examples:", style="dim")
ctx.ui.append_system(" /install-skill ./my-skill", style="dim")
ctx.ui.append_system(
" /install-skill https://github.com/user/repo/tree/main/skill-name",
style="dim",
)
ctx.ui.append_system(" /install-skill user/repo@skill-name", style="dim")
return
ctx.ui.append_system(f"Installing skill from: {source}", style="dim")
# For simplicity, calling install_skill directly (might block loop if slow?
# But install_skill doesn't seem to be async)
result = install_skill(source)
if result["success"]:
ctx.ui.append_system(f"Installed: {result['name']}", style="green")
ctx.ui.append_system(
f"Description: {result.get('description', '(none)')}",
style="dim",
)
ctx.ui.append_system(f"Path: {_shorten_path(result['path'])}", style="dim")
ctx.ui.append_system("Reload with /new to apply.", style="dim")
else:
ctx.ui.append_system(f"Failed: {result['error']}", style="red")
class InstallSkillsCommand(Command):
"""Browse and install skills."""
name = "/install-skills"
description = "Browse and install skills (optional: /install-skills <tag>)"
arguments = [
Argument(
name="tag", type=str, description="Tag to filter skills by", required=False
)
]
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from pathlib import Path as _Path
from ...paths import USER_SKILLS_DIR
from ...tools.skills_manager import fetch_remote_skill_index, install_skill
tag = args[0] if args else ""
ctx.ui.append_system(
f"Fetching skill index{' for tag: ' + tag if tag else ''}...", style="dim"
)
try:
index = fetch_remote_skill_index()
except Exception as e:
ctx.ui.append_system(f"Failed to fetch skill index: {e}", style="red")
ctx.ui.append_system(
"Try: /install-skill EvoScientist/EvoSkills@skills", style="dim"
)
return
if not index:
ctx.ui.append_system("No skills found.", style="yellow")
return
# Detect installed skills
skills_dir = _Path(USER_SKILLS_DIR)
installed_names: set[str] = set()
if skills_dir.exists():
installed_names = {e.name for e in skills_dir.iterdir() if e.is_dir()}
selected_sources: list[str] | None = None
# For non-interactive UIs (channels), if a tag is provided, we can auto-install all matching skills
# instead of failing due to lack of interactive UI.
is_channel = not ctx.ui.supports_interactive
if is_channel and tag:
tag_lower = tag.lower()
matches = []
for s in index:
tags = [t.lower() for t in s.get("tags", [])]
if tag_lower in tags:
matches.append(s)
if not matches:
ctx.ui.append_system(
f"No skills found with tag '{tag}'.", style="yellow"
)
return
ctx.ui.append_system(
f"Found {len(matches)} skill(s) with tag '{tag}'. Installing..."
)
selected_sources = [s["install_source"] for s in matches]
else:
# Wait for user interaction
selected_sources = await ctx.ui.wait_for_skill_browse(
index,
installed_names,
pre_filter_tag=tag,
)
if not selected_sources:
if not is_channel:
ctx.ui.append_system("Browse cancelled.", style="dim")
return
# Install selected skills
installed_count = 0
for source in selected_sources:
result = install_skill(source)
if result.get("batch"):
for item in result.get("installed", []):
ctx.ui.append_system(f"Installed: {item['name']}", style="green")
installed_count += 1
elif result.get("success"):
ctx.ui.append_system(f"Installed: {result['name']}", style="green")
installed_count += 1
else:
ctx.ui.append_system(
f"Failed: {result.get('error', 'unknown')}", style="red"
)
if installed_count > 0:
ctx.ui.append_system(
f"Successfully installed {installed_count} skill(s). Reload with /new to apply.",
style="dim",
)
elif not is_channel:
ctx.ui.append_system("No skills were installed.", style="yellow")
class UninstallSkillCommand(Command):
"""Remove an installed skill."""
name = "/uninstall-skill"
description = "Remove an installed skill"
arguments = [
Argument(
name="name",
type=str,
description="Name of the skill to remove",
required=True,
)
]
async def execute(self, ctx: CommandContext, args: list[str]) -> None:
from ...tools.skills_manager import uninstall_skill
name = args[0] if args else ""
if not name:
ctx.ui.append_system("Usage: /uninstall-skill <skill-name>", style="yellow")
ctx.ui.append_system("Use /skills to see installed skills.", style="dim")
return
result = uninstall_skill(name)
if result["success"]:
ctx.ui.append_system(f"Uninstalled: {name}", style="green")
ctx.ui.append_system("Reload with /new to apply.", style="dim")
else:
ctx.ui.append_system(f"Failed: {result['error']}", style="red")
# Register skill commands
manager.register(SkillsCommand())
manager.register(InstallSkillCommand())
manager.register(InstallSkillsCommand())
manager.register(UninstallSkillCommand())
EvoScientist/commands/manager.py
+88
View File
@@ -0,0 +1,88 @@
from __future__ import annotations
import logging
import shlex
from typing import Dict, List, Tuple
from .base import Command, CommandContext
_logger = logging.getLogger(__name__)
class CommandManager:
"""Manages slash command registration and execution."""
def __init__(self) -> None:
self._commands: Dict[str, Command] = {}
def register(self, command: Command) -> None:
"""Register a command and its aliases."""
names = [command.name] + command.alias
for name in names:
name = name.lower()
if not name.startswith("/"):
name = f"/{name}"
self._commands[name] = command
def get_command(self, name: str) -> Command | None:
"""Lookup a command by name."""
return self._commands.get(name.lower())
def list_commands(self) -> List[Tuple[str, str]]:
"""List all registered command names and descriptions."""
seen = set()
results = []
for cmd in self._commands.values():
if cmd not in seen:
results.append((cmd.name, cmd.description))
seen.add(cmd)
return results
def get_all_commands(self) -> List[Command]:
"""Return all registered command instances."""
seen = set()
results = []
for cmd in self._commands.values():
if cmd not in seen:
results.append(cmd)
seen.add(cmd)
return results
async def execute(self, command_str: str, ctx: CommandContext) -> bool:
"""Parse and execute a command string.
Returns True if a command was found and executed, False otherwise.
"""
command_str = command_str.strip()
if not command_str:
return False
try:
parts = shlex.split(command_str)
except ValueError:
# Fallback for unbalanced quotes
parts = command_str.split()
if not parts:
return False
cmd_name = parts[0].lower()
args = parts[1:]
cmd = self.get_command(cmd_name)
if not cmd:
return False
try:
await cmd.execute(ctx, args)
await ctx.ui.flush()
return True
except Exception as e:
_logger.exception(f"Error executing command {cmd_name}: {e}")
ctx.ui.append_system(f"Error executing {cmd_name}: {e}", style="red")
await ctx.ui.flush()
return True
# Global manager instance
manager = CommandManager()
EvoScientist/tools/skills_manager.py
+28
View File
@@ -27,6 +27,7 @@ Usage:
from __future__ import annotations
import logging
import os
import re
import shutil
@@ -40,6 +41,8 @@ import yaml
from .. import paths
_logger = logging.getLogger(__name__)
@dataclass
class SkillInfo:
@@ -278,6 +281,31 @@ def install_skill(source: str, dest_dir: str | None = None) -> dict:
if _is_github_url(source):
return _install_from_github(source, dest_dir)
else:
# Check if local path exists
source_path = Path(source).expanduser().resolve()
if not source_path.exists():
# Fallback: try resolving as a virtual workspace path
from ..paths import resolve_virtual_path
try:
if resolve_virtual_path(source).exists():
return _install_from_local(source, dest_dir)
except Exception:
pass
# If not local and not a GitHub URL, try remote lookup in EvoSkills
# This handles /install-skill skill-name shorthand
try:
index = fetch_remote_skill_index()
for skill in index:
if skill["name"].lower() == source.lower():
_logger.info(
f"Skill '{source}' found in remote index. Installing..."
)
return _install_from_github(skill["install_source"], dest_dir)
except Exception as e:
_logger.warning(f"Failed to fetch remote index for fallback: {e}")
return _install_from_local(source, dest_dir)
tests/test_channel_comprehensive.py
+32 -2
View File
@@ -292,13 +292,43 @@ class TestChunkText:
assert all(len(c) <= 50 for c in chunks)
def test_code_block_fence_split(self):
"""[B-08] Code block fence splitting should not break mid-block."""
"""[B-08] Code block fence splitting should not break mid-block without refencing."""
code = "```python\nprint('hello')\nprint('world')\n```"
text = "Before.\n\n" + code + "\n\nAfter some text here."
chunks = chunk_text(text, 40)
# Use a limit that forces a split inside the code block
chunks = chunk_text(text, 25)
# Verify we get multiple chunks and none are empty
assert len(chunks) >= 2
assert all(c.strip() for c in chunks)
# Check that the code block was properly re-fenced
# The first chunk should open the block but not close it (if it splits mid-block)
# Actually, the new implementation adds closing fences to the split part and opens the next.
# Let's just check that all chunks are valid markdown and the code is preserved.
reconstructed = "".join(c for c in chunks).replace("```python\n", "").replace("\n```", "").replace("```\n", "")
assert "print('hello')" in reconstructed
assert "print('world')" in reconstructed
# At least one chunk should have a re-fenced code block if it split
has_refence = any("```python\n" in c and c.count("```") == 2 for c in chunks)
assert has_refence, "No chunks were properly re-fenced"
# It's possible it split exactly on the fence, so we can't assert has_refence strictly without knowing the exact cut,
# but we can assert that every chunk has balanced or correctly formatted fences.
for c in chunks:
if "```" in c:
assert c.count("```") % 2 == 0, f"Unbalanced fences in chunk: {c}"
def test_long_code_block_refencing(self):
"""Verify that a very long code block is split and each chunk gets fences."""
code = "```js\n" + "line of code\n" * 10 + "```"
chunks = chunk_text(code, 50)
assert len(chunks) > 1
for chunk in chunks:
assert chunk.startswith("```js\n") or chunk.startswith("```")
assert chunk.rstrip().endswith("```")
assert chunk.count("```") >= 2
def test_code_block_preserved_when_fits(self):
code = "```\ncode\n```"
tests/test_discord_channel.py
+1 -24
View File
@@ -2,34 +2,11 @@
import pytest
from EvoScientist.channels.discord.channel import DiscordChannel, DiscordConfig
from EvoScientist.channels.base import ChannelError
from EvoScientist.channels.discord.channel import DiscordChannel, DiscordConfig
from tests.conftest import run_async as _run
class TestDiscordConfig:
def test_default_values(self):
config = DiscordConfig()
assert config.bot_token == ""
assert config.allowed_senders is None
assert config.allowed_channels is None
assert config.text_chunk_limit == 4096
def test_custom_values(self):
config = DiscordConfig(
bot_token="test-token",
allowed_senders={"111"},
allowed_channels={"222"},
text_chunk_limit=1000,
)
assert config.bot_token == "test-token"
assert config.allowed_senders == {"111"}
assert config.allowed_channels == {"222"}
assert config.text_chunk_limit == 1000
class TestDiscordChannel:
def test_init(self):
config = DiscordConfig(bot_token="test")