This commit is contained in:
X-iZhang
2026-02-03 17:23:23 +00:00
parent 51cf4a9acb
commit 7d4b8e6557
12 changed files with 1166 additions and 37 deletions
+1
View File
@@ -23,6 +23,7 @@ venv/
.github/
.cursor/
.cursorrules
.evoscientist/
# Project specific
.langgraph_api/
+17 -9
View File
@@ -25,7 +25,13 @@ from .backends import CustomSandboxBackend, MergedReadOnlyBackend
from .middleware import create_skills_middleware, create_memory_middleware
from .prompts import RESEARCHER_INSTRUCTIONS, get_system_prompt
from .utils import load_subagents
from .tools import tavily_search, think_tool
from .tools import tavily_search, think_tool, skill_manager
from .paths import (
ensure_dirs,
default_workspace_dir,
MEMORY_DIR as _MEMORY_DIR_PATH,
USER_SKILLS_DIR as _USER_SKILLS_DIR_PATH,
)
# =============================================================================
# Configuration
@@ -39,9 +45,11 @@ MAX_CONCURRENT = 3 # Max parallel sub-agents
MAX_ITERATIONS = 3 # Max delegation rounds
# Workspace settings
WORKSPACE_DIR = "./workspace/"
MEMORY_DIR = "./memory/" # Shared across sessions (not per-session)
ensure_dirs()
WORKSPACE_DIR = str(default_workspace_dir())
MEMORY_DIR = str(_MEMORY_DIR_PATH) # Shared across sessions (not per-session)
SKILLS_DIR = str(Path(__file__).parent / "skills")
USER_SKILLS_DIR = str(_USER_SKILLS_DIR_PATH)
SUBAGENTS_CONFIG = Path(__file__).parent / "subagent.yaml"
# =============================================================================
@@ -79,7 +87,7 @@ else:
# Skills backend: merge user-installed (./skills/) and system (package) skills
_skills_backend = MergedReadOnlyBackend(
primary_dir="./skills/", # user-installed, takes priority
primary_dir=USER_SKILLS_DIR, # user-installed, takes priority
secondary_dir=SKILLS_DIR, # package built-in, fallback
)
@@ -117,12 +125,12 @@ subagents = load_subagents(
_AGENT_KWARGS = dict(
name="EvoScientist",
model=chat_model,
tools=[think_tool],
tools=[think_tool, skill_manager],
backend=backend,
subagents=subagents,
middleware=[
create_memory_middleware(MEMORY_DIR, extraction_model=chat_model),
create_skills_middleware(SKILLS_DIR, "."),
create_skills_middleware(SKILLS_DIR, user_skills_dir=USER_SKILLS_DIR),
],
system_prompt=SYSTEM_PROMPT,
)
@@ -137,7 +145,7 @@ def create_cli_agent(workspace_dir: str | None = None):
Args:
workspace_dir: Optional per-session workspace directory. If provided,
creates a fresh backend rooted at this path. If None, uses the
module-level default backend (./workspace/).
module-level default backend (./workspace).
"""
from langgraph.checkpoint.memory import InMemorySaver # type: ignore[import-untyped]
@@ -148,7 +156,7 @@ def create_cli_agent(workspace_dir: str | None = None):
timeout=300,
)
sk_backend = MergedReadOnlyBackend(
primary_dir="./skills/",
primary_dir=USER_SKILLS_DIR,
secondary_dir=SKILLS_DIR,
)
# Memory always uses SHARED directory (not per-session) for cross-session persistence
@@ -165,7 +173,7 @@ def create_cli_agent(workspace_dir: str | None = None):
)
mw = [
create_memory_middleware(MEMORY_DIR, extraction_model=chat_model),
create_skills_middleware(SKILLS_DIR, "."),
create_skills_middleware(SKILLS_DIR, user_skills_dir=USER_SKILLS_DIR),
]
kwargs = dict(
_AGENT_KWARGS,
+190 -16
View File
@@ -12,6 +12,7 @@ Features:
- Interactive mode with prompt_toolkit
"""
import logging
import os
import sys
import uuid
@@ -28,6 +29,23 @@ from rich.text import Text # type: ignore[import-untyped]
# Backward-compat re-exports (tests import these from EvoScientist.cli)
from .stream.state import SubAgentState, StreamState, _parse_todo_items, _build_todo_stats # noqa: F401
from .stream.display import console, _run_streaming
from .paths import ensure_dirs, new_run_dir
def _shorten_path(path: str) -> str:
"""Shorten absolute path to relative path from current directory."""
if not path:
return path
try:
cwd = os.getcwd()
if path.startswith(cwd):
# Remove cwd prefix, keep the relative part
rel = path[len(cwd):].lstrip(os.sep)
# Add current dir name for context
return os.path.join(os.path.basename(cwd), rel) if rel else os.path.basename(cwd)
return path
except Exception:
return path
# =============================================================================
@@ -60,33 +78,118 @@ def print_banner(
info.append(thread_id, style="yellow")
if workspace_dir:
info.append("\n Workspace: ", style="dim")
info.append(workspace_dir, style="cyan")
info.append(_shorten_path(workspace_dir), style="cyan")
if memory_dir:
trimmed = memory_dir.rstrip("/").rstrip("\\")
info.append("\n Memory dir: ", style="dim")
info.append(trimmed, style="cyan")
info.append(_shorten_path(trimmed), style="cyan")
info.append("\n Commands: ", style="dim")
info.append("/exit", style="bold")
info.append(", ", style="dim")
info.append("/new", style="bold")
info.append(" (new session), ", style="dim")
info.append(", ", style="dim")
info.append("/thread", style="bold")
info.append(" (show thread ID)", style="dim")
info.append(", ", style="dim")
info.append("/skills", style="bold")
info.append(", ", style="dim")
info.append("/install-skill", style="bold")
info.append(", ", style="dim")
info.append("/uninstall-skill", style="bold")
console.print(info)
console.print()
# =============================================================================
# Skill management commands
# =============================================================================
def _cmd_list_skills() -> None:
"""List installed user skills."""
from .skills_manager import list_skills
from .paths import USER_SKILLS_DIR
skills = list_skills(include_system=False)
if not skills:
console.print("[dim]No user skills installed.[/dim]")
console.print(f"[dim]Install with:[/dim] /install-skill <path-or-url>")
console.print(f"[dim]Skills directory:[/dim] [cyan]{_shorten_path(str(USER_SKILLS_DIR))}[/cyan]")
console.print()
return
console.print(f"[bold]Installed Skills[/bold] ({len(skills)}):")
for skill in skills:
console.print(f" [green]{skill.name}[/green] - {skill.description}")
console.print(f"\n[dim]Location:[/dim] [cyan]{_shorten_path(str(USER_SKILLS_DIR))}[/cyan]")
console.print()
def _cmd_install_skill(source: str) -> None:
"""Install a skill from local path or GitHub URL."""
from .skills_manager import install_skill
if not source:
console.print("[red]Usage:[/red] /install-skill <path-or-url>")
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()
return
console.print(f"[dim]Installing skill from:[/dim] {source}")
result = install_skill(source)
if 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 the agent with /new to use the skill.[/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 .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 the agent with /new to apply changes.[/dim]")
else:
console.print(f"[red]Failed:[/red] {result['error']}")
console.print()
# =============================================================================
# CLI commands
# =============================================================================
def cmd_interactive(agent: Any, show_thinking: bool = True, workspace_dir: str | None = None) -> None:
def cmd_interactive(
agent: Any,
show_thinking: bool = True,
workspace_dir: str | None = None,
workspace_fixed: bool = False,
) -> None:
"""Interactive conversation mode with streaming output.
Args:
agent: Compiled agent graph
show_thinking: Whether to display thinking panels
workspace_dir: Per-session workspace directory path
workspace_fixed: If True, /new keeps the same workspace directory
"""
thread_id = str(uuid.uuid4())
from .EvoScientist import MEMORY_DIR
@@ -126,24 +229,40 @@ def cmd_interactive(agent: Any, show_thinking: bool = True, workspace_dir: str |
break
if user_input.lower() == "/new":
# New session: new workspace, new agent, new thread
workspace_dir = _create_session_workspace()
# New session: new thread; workspace only changes if not fixed
if not workspace_fixed:
workspace_dir = _create_session_workspace()
console.print("[dim]Loading new session...[/dim]")
agent = _load_agent(workspace_dir=workspace_dir)
thread_id = str(uuid.uuid4())
console.print(f"[green]New session:[/green] [yellow]{thread_id}[/yellow]")
console.print(f"[dim]Workspace:[/dim] [cyan]{workspace_dir}[/cyan]\n")
if workspace_dir:
console.print(f"[dim]Workspace:[/dim] [cyan]{_shorten_path(workspace_dir)}[/cyan]\n")
continue
if user_input.lower() == "/thread":
console.print(f"[dim]Thread:[/dim] [yellow]{thread_id}[/yellow]")
if workspace_dir:
console.print(f"[dim]Workspace:[/dim] [cyan]{workspace_dir}[/cyan]")
console.print(f"[dim]Workspace:[/dim] [cyan]{_shorten_path(workspace_dir)}[/cyan]")
if memory_dir:
console.print(f"[dim]Memory dir:[/dim] [cyan]{memory_dir}[/cyan]")
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
# Stream agent response
console.print()
_run_streaming(agent, user_input, thread_id, show_thinking, interactive=True)
@@ -175,7 +294,7 @@ def cmd_run(agent: Any, prompt: str, thread_id: str | None = None, show_thinking
console.print(sep)
console.print(f"[dim]Thread: {thread_id}[/dim]")
if workspace_dir:
console.print(f"[dim]Workspace: {workspace_dir}[/dim]")
console.print(f"[dim]Workspace: {_shorten_path(workspace_dir)}[/dim]")
console.print()
try:
@@ -192,7 +311,7 @@ def cmd_run(agent: Any, prompt: str, thread_id: str | None = None, show_thinking
def _create_session_workspace() -> str:
"""Create a per-session workspace directory and return its path."""
session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
workspace_dir = os.path.join(".", "workspace", session_id)
workspace_dir = str(new_run_dir(session_id))
os.makedirs(workspace_dir, exist_ok=True)
return workspace_dir
@@ -221,6 +340,8 @@ def _main_callback(
interactive: bool = typer.Option(False, "-i", "--interactive", help="Interactive conversation mode"),
thread_id: Optional[str] = typer.Option(None, "--thread-id", help="Thread ID for conversation persistence"),
no_thinking: bool = typer.Option(False, "--no-thinking", help="Disable thinking display"),
workdir: Optional[str] = typer.Option(None, "--workdir", help="Override workspace directory for this session"),
use_cwd: bool = typer.Option(False, "--use-cwd", help="Use current working directory as workspace"),
):
"""EvoScientist Agent - AI-powered research & code execution CLI."""
from dotenv import load_dotenv # type: ignore[import-untyped]
@@ -228,24 +349,77 @@ def _main_callback(
show_thinking = not no_thinking
# Create per-session workspace
workspace_dir = _create_session_workspace()
if workdir and use_cwd:
raise typer.BadParameter("Use either --workdir or --use-cwd, not both.")
ensure_dirs()
# Resolve workspace directory for this session
if use_cwd:
workspace_dir = os.getcwd()
workspace_fixed = True
elif workdir:
workspace_dir = os.path.abspath(os.path.expanduser(workdir))
os.makedirs(workspace_dir, exist_ok=True)
workspace_fixed = True
else:
workspace_dir = _create_session_workspace()
workspace_fixed = False
# Load agent with session workspace
console.print("[dim]Loading agent...[/dim]")
agent = _load_agent(workspace_dir=workspace_dir)
if interactive:
cmd_interactive(agent, show_thinking=show_thinking, workspace_dir=workspace_dir)
cmd_interactive(
agent,
show_thinking=show_thinking,
workspace_dir=workspace_dir,
workspace_fixed=workspace_fixed,
)
elif prompt:
cmd_run(agent, prompt, thread_id=thread_id, show_thinking=show_thinking, workspace_dir=workspace_dir)
else:
# Default: interactive mode
cmd_interactive(agent, show_thinking=show_thinking, workspace_dir=workspace_dir)
cmd_interactive(
agent,
show_thinking=show_thinking,
workspace_dir=workspace_dir,
workspace_fixed=workspace_fixed,
)
def _configure_logging():
"""Configure logging with warning symbols for better visibility."""
from rich.logging import RichHandler
class DimWarningHandler(RichHandler):
"""Custom handler that renders warnings in dim style."""
def emit(self, record: logging.LogRecord) -> None:
if record.levelno == logging.WARNING:
# Use Rich console to print dim warning
msg = record.getMessage()
console.print(f"[dim yellow]\u26a0\ufe0f Warning:[/dim yellow] [dim]{msg}[/dim]")
else:
super().emit(record)
# Configure root logger to use our handler for WARNING and above
handler = DimWarningHandler(console=console, show_time=False, show_path=False, show_level=False)
handler.setLevel(logging.WARNING)
# Apply to root logger (catches all loggers including deepagents)
root_logger = logging.getLogger()
# Remove existing handlers to avoid duplicate output
for h in root_logger.handlers[:]:
root_logger.removeHandler(h)
root_logger.addHandler(handler)
root_logger.setLevel(logging.WARNING)
def main():
"""CLI entry point — delegates to the Typer app."""
_configure_logging()
app()
+8 -2
View File
@@ -10,6 +10,7 @@ from deepagents.middleware.skills import SkillsMiddleware
from .backends import MergedReadOnlyBackend
from .memory import EvoMemoryMiddleware
from .paths import MEMORY_DIR as _DEFAULT_MEMORY_DIR
if TYPE_CHECKING:
from langchain.chat_models import BaseChatModel
@@ -20,6 +21,7 @@ _DEFAULT_SKILLS_DIR = str(Path(__file__).parent / "skills")
def create_skills_middleware(
skills_dir: str = _DEFAULT_SKILLS_DIR,
workspace_dir: str = ".",
user_skills_dir: str | None = None,
) -> SkillsMiddleware:
"""Create a SkillsMiddleware that loads skills.
@@ -29,12 +31,16 @@ def create_skills_middleware(
Args:
skills_dir: Path to the system skills directory (package built-in)
workspace_dir: Path to the project root (user skills live under {workspace_dir}/skills/)
user_skills_dir: Optional explicit path for user-installed skills. If set,
this path is used directly instead of {workspace_dir}/skills.
Returns:
Configured SkillsMiddleware instance
"""
if user_skills_dir is None:
user_skills_dir = str(Path(workspace_dir) / "skills")
merged = MergedReadOnlyBackend(
primary_dir=str(Path(workspace_dir) / "skills"),
primary_dir=user_skills_dir,
secondary_dir=skills_dir,
)
return SkillsMiddleware(
@@ -44,7 +50,7 @@ def create_skills_middleware(
def create_memory_middleware(
memory_dir: str = "./memory/",
memory_dir: str = str(_DEFAULT_MEMORY_DIR),
extraction_model: BaseChatModel | None = None,
trigger: tuple[str, int] = ("messages", 20),
) -> EvoMemoryMiddleware:
+45
View File
@@ -0,0 +1,45 @@
"""Path resolution utilities for EvoScientist runtime directories."""
from __future__ import annotations
import os
from datetime import datetime
from pathlib import Path
def _expand(path: str) -> Path:
return Path(path).expanduser()
def _env_path(key: str) -> Path | None:
value = os.getenv(key)
if not value:
return None
return _expand(value)
STATE_ROOT = _env_path("EVOSCIENTIST_HOME") or (Path.cwd() / ".evoscientist")
WORKSPACE_ROOT = _env_path("EVOSCIENTIST_WORKSPACE_DIR") or (STATE_ROOT / "workspace")
RUNS_DIR = _env_path("EVOSCIENTIST_RUNS_DIR") or (WORKSPACE_ROOT / "runs")
MEMORY_DIR = _env_path("EVOSCIENTIST_MEMORY_DIR") or (WORKSPACE_ROOT / "memory")
USER_SKILLS_DIR = _env_path("EVOSCIENTIST_SKILLS_DIR") or (WORKSPACE_ROOT / "skills")
def ensure_dirs() -> None:
"""Create runtime directories if they do not exist."""
for path in (WORKSPACE_ROOT, RUNS_DIR, MEMORY_DIR, USER_SKILLS_DIR):
path.mkdir(parents=True, exist_ok=True)
def default_workspace_dir() -> Path:
"""Default workspace for non-CLI usage."""
return WORKSPACE_ROOT
def new_run_dir(session_id: str | None = None) -> Path:
"""Create a new run directory name under RUNS_DIR (path only)."""
if session_id is None:
session_id = datetime.now().strftime("%Y%m%d_%H%M%S")
return RUNS_DIR / session_id
+392
View File
@@ -0,0 +1,392 @@
"""Skill installation and management for EvoScientist.
This module provides functions for installing, listing, and uninstalling user skills.
Skills are installed to USER_SKILLS_DIR (~/.evoscientist/workspace/skills/).
Supported installation sources:
- Local directory paths
- GitHub URLs (https://github.com/owner/repo or .../tree/branch/path)
- GitHub shorthand (owner/repo@skill-name)
Usage:
from EvoScientist.skills_manager import install_skill, list_skills, uninstall_skill
# Install from local path
install_skill("./my-skill")
# Install from GitHub
install_skill("https://github.com/user/repo/tree/main/my-skill")
# List installed skills
for skill in list_skills():
print(skill["name"], skill["description"])
# Uninstall a skill
uninstall_skill("my-skill")
"""
from __future__ import annotations
import os
import re
import shutil
import subprocess
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
import yaml
from .paths import USER_SKILLS_DIR
@dataclass
class SkillInfo:
"""Information about an installed skill."""
name: str
description: str
path: Path
source: str # "user" or "system"
def _parse_skill_md(skill_md_path: Path) -> dict[str, str]:
"""Parse SKILL.md frontmatter to extract name and description.
SKILL.md format:
---
name: skill-name
description: A brief description...
---
# Skill Title
...
Returns:
Dictionary with 'name' and 'description' keys.
"""
content = skill_md_path.read_text(encoding="utf-8")
# Extract YAML frontmatter
frontmatter_match = re.match(r"^---\s*\n(.*?)\n---", content, re.DOTALL)
if not frontmatter_match:
# No frontmatter, use directory name
return {
"name": skill_md_path.parent.name,
"description": "(no description)",
}
try:
frontmatter = yaml.safe_load(frontmatter_match.group(1))
return {
"name": frontmatter.get("name", skill_md_path.parent.name),
"description": frontmatter.get("description", "(no description)"),
}
except yaml.YAMLError:
return {
"name": skill_md_path.parent.name,
"description": "(invalid frontmatter)",
}
def _parse_github_url(url: str) -> tuple[str, str | None, str | None]:
"""Parse a GitHub URL into (repo, ref, path).
Supports formats:
https://github.com/owner/repo
https://github.com/owner/repo/tree/main/path/to/skill
github.com/owner/repo/tree/branch/path
owner/repo@skill-name (shorthand from skills.sh)
Returns:
(repo, ref_or_none, path_or_none)
"""
# Shorthand: owner/repo@path
if "@" in url and "://" not in url:
repo, path = url.split("@", 1)
return repo.strip(), None, path.strip()
# Strip protocol and github.com prefix
cleaned = re.sub(r"^https?://", "", url)
cleaned = re.sub(r"^github\.com/", "", cleaned)
cleaned = cleaned.rstrip("/")
# Match: owner/repo/tree/ref/path...
m = re.match(r"^([^/]+/[^/]+)/tree/([^/]+)(?:/(.+))?$", cleaned)
if m:
return m.group(1), m.group(2), m.group(3)
# Match: owner/repo (no tree)
m = re.match(r"^([^/]+/[^/]+)$", cleaned)
if m:
return m.group(1), None, None
raise ValueError(f"Cannot parse GitHub URL: {url}")
def _clone_repo(repo: str, ref: str | None, dest: str) -> None:
"""Shallow-clone a GitHub repo."""
clone_url = f"https://github.com/{repo}.git"
cmd = ["git", "clone", "--depth", "1"]
if ref:
cmd += ["--branch", ref]
cmd += [clone_url, dest]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"git clone failed: {result.stderr.strip()}")
def _is_github_url(source: str) -> bool:
"""Check if the source looks like a GitHub URL or shorthand."""
if "github.com" in source.lower():
return True
if "://" in source:
return False # Non-GitHub URL
# Check for owner/repo@skill shorthand
if "@" in source and "/" in source.split("@")[0]:
return True
# Check for owner/repo format (but not local paths like ./foo or /foo)
if "/" in source and not source.startswith((".", "/")):
parts = source.split("/")
# GitHub shorthand: exactly 2 parts, both non-empty, no extensions
if len(parts) == 2 and all(parts) and "." not in parts[0]:
return True
return False
def _validate_skill_dir(path: Path) -> bool:
"""Check if a directory contains a valid skill (has SKILL.md)."""
return (path / "SKILL.md").is_file()
def install_skill(source: str, dest_dir: str | None = None) -> dict:
"""Install a skill from a local path or GitHub URL.
Args:
source: Local directory path or GitHub URL/shorthand.
dest_dir: Destination directory (defaults to USER_SKILLS_DIR).
Returns:
Dictionary with installation result:
- success: bool
- name: skill name (if successful)
- path: installed path (if successful)
- error: error message (if failed)
"""
dest_dir = dest_dir or str(USER_SKILLS_DIR)
os.makedirs(dest_dir, exist_ok=True)
if _is_github_url(source):
return _install_from_github(source, dest_dir)
else:
return _install_from_local(source, dest_dir)
def _install_from_local(source: str, dest_dir: str) -> dict:
"""Install a skill from a local directory path."""
source_path = Path(source).expanduser().resolve()
if not source_path.exists():
return {"success": False, "error": f"Path does not exist: {source}"}
if not source_path.is_dir():
return {"success": False, "error": f"Not a directory: {source}"}
if not _validate_skill_dir(source_path):
return {"success": False, "error": f"No SKILL.md found in: {source}"}
# Parse SKILL.md to get the skill name
skill_info = _parse_skill_md(source_path / "SKILL.md")
skill_name = skill_info["name"]
# Destination path
target_path = Path(dest_dir) / skill_name
# Remove existing if present
if target_path.exists():
shutil.rmtree(target_path)
# Copy skill directory
shutil.copytree(source_path, target_path)
return {
"success": True,
"name": skill_name,
"path": str(target_path),
"description": skill_info["description"],
}
def _install_from_github(source: str, dest_dir: str) -> dict:
"""Install a skill from a GitHub URL or shorthand."""
try:
repo, ref, path = _parse_github_url(source)
except ValueError as e:
return {"success": False, "error": str(e)}
with tempfile.TemporaryDirectory(prefix="evoscientist-skill-") as tmp:
clone_dir = os.path.join(tmp, "repo")
try:
_clone_repo(repo, ref, clone_dir)
except RuntimeError as e:
return {"success": False, "error": str(e)}
# Determine the skill source directory
if path:
skill_source = Path(clone_dir) / path
else:
skill_source = Path(clone_dir)
if not skill_source.exists():
return {"success": False, "error": f"Path not found in repo: {path or '/'}"}
if not _validate_skill_dir(skill_source):
# Maybe the repo root contains multiple skills?
if not path:
# Try to find skills in repo root
found_skills = []
for entry in os.listdir(clone_dir):
entry_path = Path(clone_dir) / entry
if entry_path.is_dir() and _validate_skill_dir(entry_path):
found_skills.append(entry)
if found_skills:
return {
"success": False,
"error": (
f"Multiple skills found in repo. "
f"Please specify one: {', '.join(found_skills)}"
),
}
return {"success": False, "error": f"No SKILL.md found in: {source}"}
# Parse skill info and copy
skill_info = _parse_skill_md(skill_source / "SKILL.md")
skill_name = skill_info["name"]
target_path = Path(dest_dir) / skill_name
if target_path.exists():
shutil.rmtree(target_path)
# Copy, excluding .git directory
def ignore_git(dir_name: str, files: list[str]) -> list[str]:
return [f for f in files if f == ".git"]
shutil.copytree(skill_source, target_path, ignore=ignore_git)
return {
"success": True,
"name": skill_name,
"path": str(target_path),
"description": skill_info["description"],
"source": source,
}
def list_skills(include_system: bool = False) -> list[SkillInfo]:
"""List all installed user skills.
Args:
include_system: If True, also include system (built-in) skills.
Returns:
List of SkillInfo objects for each installed skill.
"""
skills: list[SkillInfo] = []
# User skills
user_dir = Path(USER_SKILLS_DIR)
if user_dir.exists():
for entry in sorted(user_dir.iterdir()):
if entry.is_dir() and _validate_skill_dir(entry):
skill_md = entry / "SKILL.md"
info = _parse_skill_md(skill_md)
skills.append(
SkillInfo(
name=info["name"],
description=info["description"],
path=entry,
source="user",
)
)
# System skills (optional)
if include_system:
from .EvoScientist import SKILLS_DIR
system_dir = Path(SKILLS_DIR)
if system_dir.exists():
for entry in sorted(system_dir.iterdir()):
if entry.is_dir() and _validate_skill_dir(entry):
# Skip if user has overridden this skill
if any(s.name == entry.name for s in skills):
continue
skill_md = entry / "SKILL.md"
info = _parse_skill_md(skill_md)
skills.append(
SkillInfo(
name=info["name"],
description=info["description"],
path=entry,
source="system",
)
)
return skills
def uninstall_skill(name: str) -> dict:
"""Uninstall a user-installed skill.
Args:
name: Name of the skill to uninstall.
Returns:
Dictionary with result:
- success: bool
- error: error message (if failed)
"""
user_dir = Path(USER_SKILLS_DIR)
target_path = user_dir / name
if not target_path.exists():
# Try to find by directory name (in case name differs from dir name)
found = None
if user_dir.exists():
for entry in user_dir.iterdir():
if entry.is_dir() and _validate_skill_dir(entry):
info = _parse_skill_md(entry / "SKILL.md")
if info["name"] == name:
found = entry
break
if not found:
return {"success": False, "error": f"Skill not found: {name}"}
target_path = found
# Check if it's a user skill (not system)
if not str(target_path).startswith(str(user_dir)):
return {"success": False, "error": f"Cannot uninstall system skill: {name}"}
# Remove the skill directory
shutil.rmtree(target_path)
return {"success": True, "name": name}
def get_skill_info(name: str) -> SkillInfo | None:
"""Get information about a specific skill.
Args:
name: Name of the skill.
Returns:
SkillInfo if found, None otherwise.
"""
for skill in list_skills(include_system=True):
if skill.name == name:
return skill
return None
+15
View File
@@ -141,7 +141,22 @@ def _render_tool_call_line(tc: dict, tr: dict | None) -> Text:
style = "bold yellow" if not is_task else "bold cyan"
indicator = "\u25b6" if is_task else ToolStatus.RUNNING.value
# Try to get display name from args first
tool_compact = format_tool_compact(tc['name'], tc.get('args'))
# If args were empty and we have a result, try to infer memory operations from result
tool_name = tc.get('name', '').lower()
if tool_name in ('write_file', 'edit_file') and tr is not None:
result_content = tr.get('content', '')
if '/MEMORY.md' in result_content or 'MEMORY.md' in result_content:
tool_compact = "Updating memory"
elif tool_name == 'read_file' and tr is not None:
result_content = tr.get('content', '')
# read_file result doesn't contain path, check if args is empty and result looks like memory
args = tc.get('args') or {}
if not args.get('path') and '# EvoScientist Memory' in result_content:
tool_compact = "Reading memory"
tool_text = Text()
tool_text.append(f"{indicator} ", style=style)
tool_text.append(tool_compact, style=style)
+13 -7
View File
@@ -117,18 +117,24 @@ def format_tool_compact(name: str, args: dict | None) -> str:
cmd = cmd[:47] + "\u2026"
return f"execute({cmd})"
# File operations
# File operations (with special case for memory files)
if name_lower == "read_file":
path = _shorten_path(args.get("path", ""))
return f"read_file({path})"
path = args.get("path", "")
if path.endswith("/MEMORY.md") or path == "/MEMORY.md":
return "Reading memory"
return f"read_file({_shorten_path(path)})"
if name_lower == "write_file":
path = _shorten_path(args.get("path", ""))
return f"write_file({path})"
path = args.get("path", "")
if path.endswith("/MEMORY.md") or path == "/MEMORY.md":
return "Updating memory"
return f"write_file({_shorten_path(path)})"
if name_lower == "edit_file":
path = _shorten_path(args.get("path", ""))
return f"edit_file({path})"
path = args.get("path", "")
if path.endswith("/MEMORY.md") or path == "/MEMORY.md":
return "Updating memory"
return f"edit_file({_shorten_path(path)})"
# Search operations
if name_lower == "glob":
+64
View File
@@ -106,6 +106,70 @@ async def tavily_search(
return f"Search failed: {str(e)}"
@tool(parse_docstring=True)
def skill_manager(
action: Literal["install", "list", "uninstall"],
source: str = "",
name: str = "",
) -> str:
"""Manage user skills: install, list, or uninstall.
Use this tool when the user asks to:
- Install a skill (action="install", source required)
- List installed skills (action="list")
- Uninstall a skill (action="uninstall", name required)
Supported sources for install:
- Local path: "./my-skill" or "/path/to/skill"
- GitHub URL: "https://github.com/owner/repo/tree/main/skill-name"
- GitHub shorthand: "owner/repo@skill-name"
Args:
action: One of "install", "list", or "uninstall"
source: For install - local path or GitHub URL/shorthand
name: For uninstall - skill name to remove
Returns:
Result message
"""
from .skills_manager import install_skill, list_skills, uninstall_skill
if action == "install":
if not source:
return "Error: 'source' is required for install action"
result = install_skill(source)
if result["success"]:
return (
f"Successfully installed skill: {result['name']}\n"
f"Description: {result.get('description', '(none)')}\n"
f"Path: {result['path']}\n\n"
f"Use load_skill to activate it."
)
else:
return f"Failed to install skill: {result['error']}"
elif action == "list":
skills = list_skills(include_system=False)
if not skills:
return "No user skills installed. Use action='install' to add skills."
lines = [f"Installed User Skills ({len(skills)}):"]
for skill in skills:
lines.append(f" - {skill.name}: {skill.description}")
return "\n".join(lines)
elif action == "uninstall":
if not name:
return "Error: 'name' is required for uninstall action"
result = uninstall_skill(name)
if result["success"]:
return f"Successfully uninstalled skill: {name}"
else:
return f"Failed to uninstall skill: {result['error']}"
else:
return f"Unknown action: {action}. Use 'install', 'list', or 'uninstall'."
@tool(parse_docstring=True)
def think_tool(reflection: str) -> str:
"""Tool for strategic reflection on research progress and decision-making.
+59 -3
View File
@@ -125,9 +125,65 @@ or
```Shell
EvoSci # or EvoScientist
```
**Optional arguments:**
**Optional arguments:**
> TODO
```
--workdir <path> Override workspace directory for this session
--use-cwd Use current working directory as workspace
--thread-id <id> Resume a conversation thread
--no-thinking Disable thinking display
```
**Interactive Commands:**
| Command | Description |
|---------|-------------|
| `/exit` | Quit the session |
| `/new` | Start a new session (new workspace + thread) |
| `/thread` | Show current thread ID and workspace path |
| `/skills` | List installed user skills |
| `/install-skill <source>` | Install a skill from local path or GitHub |
| `/uninstall-skill <name>` | Uninstall a user-installed skill |
**Skill Installation Examples:**
```bash
# Install from local path
/install-skill ./my-skill
# Install from GitHub URL
/install-skill https://github.com/owner/repo/tree/main/skill-name
# Install from GitHub shorthand
/install-skill owner/repo@skill-name
```
### Runtime Directories
By default, the **workspace** is created under a hidden directory in the current
project directory:
```
./.evoscientist/workspace/
memory/ # shared MEMORY.md
skills/ # user-installed skills
runs/ # per-thread workspaces
```
You can force workspace to be the current directory via `--use-cwd`.
If you set `EVOSCIENTIST_HOME`, EvoScientist will place `workspace/` under that
directory instead of the project root:
Example with `EVOSCIENTIST_HOME=~/.evoscientist`:
```
~/.evoscientist/
workspace/
memory/
skills/
runs/
```
### Script Inference
```python
@@ -229,4 +285,4 @@ For any enquiries or collaboration opportunities, please contact: [**youganglyu@
## 📜 License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
+345
View File
@@ -0,0 +1,345 @@
"""Tests for EvoScientist.skills_manager module."""
import os
import shutil
import tempfile
from pathlib import Path
from unittest import mock
import pytest
from EvoScientist.skills_manager import (
SkillInfo,
_parse_skill_md,
_parse_github_url,
_is_github_url,
_validate_skill_dir,
install_skill,
list_skills,
uninstall_skill,
)
# =============================================================================
# Fixtures
# =============================================================================
@pytest.fixture
def temp_skills_dir(tmp_path):
"""Create a temporary skills directory."""
skills_dir = tmp_path / "skills"
skills_dir.mkdir()
return skills_dir
@pytest.fixture
def sample_skill_dir(tmp_path):
"""Create a sample skill directory with SKILL.md."""
skill_dir = tmp_path / "sample-skill"
skill_dir.mkdir()
skill_md = skill_dir / "SKILL.md"
skill_md.write_text(
"""---
name: sample-skill
description: A sample skill for testing
---
# Sample Skill
This is a sample skill for testing purposes.
"""
)
return skill_dir
@pytest.fixture
def sample_skill_no_frontmatter(tmp_path):
"""Create a skill directory without YAML frontmatter."""
skill_dir = tmp_path / "no-frontmatter-skill"
skill_dir.mkdir()
skill_md = skill_dir / "SKILL.md"
skill_md.write_text(
"""# No Frontmatter Skill
This skill has no YAML frontmatter.
"""
)
return skill_dir
# =============================================================================
# Tests for _parse_skill_md
# =============================================================================
class TestParseSkillMd:
"""Tests for _parse_skill_md function."""
def test_parse_with_frontmatter(self, sample_skill_dir):
skill_md = sample_skill_dir / "SKILL.md"
result = _parse_skill_md(skill_md)
assert result["name"] == "sample-skill"
assert result["description"] == "A sample skill for testing"
def test_parse_without_frontmatter(self, sample_skill_no_frontmatter):
skill_md = sample_skill_no_frontmatter / "SKILL.md"
result = _parse_skill_md(skill_md)
# Should use directory name as fallback
assert result["name"] == "no-frontmatter-skill"
assert result["description"] == "(no description)"
def test_parse_with_partial_frontmatter(self, tmp_path):
skill_dir = tmp_path / "partial-skill"
skill_dir.mkdir()
skill_md = skill_dir / "SKILL.md"
skill_md.write_text(
"""---
name: my-skill
---
# My Skill
"""
)
result = _parse_skill_md(skill_md)
assert result["name"] == "my-skill"
assert result["description"] == "(no description)"
# =============================================================================
# Tests for _parse_github_url
# =============================================================================
class TestParseGithubUrl:
"""Tests for _parse_github_url function."""
def test_parse_full_url_with_path(self):
url = "https://github.com/owner/repo/tree/main/my-skill"
repo, ref, path = _parse_github_url(url)
assert repo == "owner/repo"
assert ref == "main"
assert path == "my-skill"
def test_parse_full_url_without_path(self):
url = "https://github.com/owner/repo/tree/develop"
repo, ref, path = _parse_github_url(url)
assert repo == "owner/repo"
assert ref == "develop"
assert path is None
def test_parse_simple_repo_url(self):
url = "https://github.com/owner/repo"
repo, ref, path = _parse_github_url(url)
assert repo == "owner/repo"
assert ref is None
assert path is None
def test_parse_shorthand(self):
url = "owner/repo@my-skill"
repo, ref, path = _parse_github_url(url)
assert repo == "owner/repo"
assert ref is None
assert path == "my-skill"
def test_parse_url_without_protocol(self):
url = "github.com/owner/repo/tree/v1.0/path/to/skill"
repo, ref, path = _parse_github_url(url)
assert repo == "owner/repo"
assert ref == "v1.0"
assert path == "path/to/skill"
def test_invalid_url_raises(self):
with pytest.raises(ValueError, match="Cannot parse"):
_parse_github_url("not-a-valid-url")
# =============================================================================
# Tests for _is_github_url
# =============================================================================
class TestIsGithubUrl:
"""Tests for _is_github_url function."""
def test_github_com_url(self):
assert _is_github_url("https://github.com/owner/repo") is True
assert _is_github_url("http://github.com/owner/repo/tree/main/skill") is True
def test_shorthand(self):
assert _is_github_url("owner/repo@skill-name") is True
def test_local_path(self):
assert _is_github_url("./my-skill") is False
assert _is_github_url("/absolute/path/skill") is False
assert _is_github_url("../relative/path") is False
def test_other_urls(self):
assert _is_github_url("https://gitlab.com/owner/repo") is False
assert _is_github_url("file:///path/to/file") is False
# =============================================================================
# Tests for _validate_skill_dir
# =============================================================================
class TestValidateSkillDir:
"""Tests for _validate_skill_dir function."""
def test_valid_skill_dir(self, sample_skill_dir):
assert _validate_skill_dir(sample_skill_dir) is True
def test_invalid_skill_dir_no_skillmd(self, tmp_path):
empty_dir = tmp_path / "empty"
empty_dir.mkdir()
assert _validate_skill_dir(empty_dir) is False
def test_invalid_skill_dir_file_not_dir(self, tmp_path):
file_path = tmp_path / "file.txt"
file_path.write_text("not a directory")
assert _validate_skill_dir(file_path) is False
# =============================================================================
# Tests for install_skill
# =============================================================================
class TestInstallSkill:
"""Tests for install_skill function."""
def test_install_from_local_path(self, sample_skill_dir, temp_skills_dir):
result = install_skill(str(sample_skill_dir), str(temp_skills_dir))
assert result["success"] is True
assert result["name"] == "sample-skill"
assert "sample-skill" in result["path"]
# Verify the skill was copied
installed_path = Path(result["path"])
assert installed_path.exists()
assert (installed_path / "SKILL.md").exists()
def test_install_nonexistent_path(self, temp_skills_dir):
result = install_skill("/nonexistent/path", str(temp_skills_dir))
assert result["success"] is False
assert "does not exist" in result["error"]
def test_install_invalid_skill_no_skillmd(self, tmp_path, temp_skills_dir):
empty_dir = tmp_path / "empty-skill"
empty_dir.mkdir()
result = install_skill(str(empty_dir), str(temp_skills_dir))
assert result["success"] is False
assert "No SKILL.md" in result["error"]
def test_install_replaces_existing(self, sample_skill_dir, temp_skills_dir):
# Install first time
result1 = install_skill(str(sample_skill_dir), str(temp_skills_dir))
assert result1["success"] is True
# Modify the original skill
skill_md = sample_skill_dir / "SKILL.md"
skill_md.write_text(
"""---
name: sample-skill
description: Modified description
---
# Modified
"""
)
# Install again
result2 = install_skill(str(sample_skill_dir), str(temp_skills_dir))
assert result2["success"] is True
assert result2["description"] == "Modified description"
# =============================================================================
# Tests for list_skills
# =============================================================================
class TestListSkills:
"""Tests for list_skills function."""
def test_list_empty_dir(self, temp_skills_dir):
with mock.patch("EvoScientist.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
skills = list_skills(include_system=False)
assert skills == []
def test_list_with_skills(self, sample_skill_dir, temp_skills_dir):
# Install a skill
install_skill(str(sample_skill_dir), str(temp_skills_dir))
with mock.patch("EvoScientist.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
skills = list_skills(include_system=False)
assert len(skills) == 1
assert skills[0].name == "sample-skill"
assert skills[0].description == "A sample skill for testing"
assert skills[0].source == "user"
def test_list_multiple_skills(self, tmp_path, temp_skills_dir):
# Create and install multiple skills
for i in range(3):
skill_dir = tmp_path / f"skill-{i}"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
f"""---
name: skill-{i}
description: Skill number {i}
---
"""
)
install_skill(str(skill_dir), str(temp_skills_dir))
with mock.patch("EvoScientist.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
skills = list_skills(include_system=False)
assert len(skills) == 3
names = [s.name for s in skills]
assert "skill-0" in names
assert "skill-1" in names
assert "skill-2" in names
# =============================================================================
# Tests for uninstall_skill
# =============================================================================
class TestUninstallSkill:
"""Tests for uninstall_skill function."""
def test_uninstall_existing_skill(self, sample_skill_dir, temp_skills_dir):
# Install first
install_skill(str(sample_skill_dir), str(temp_skills_dir))
with mock.patch("EvoScientist.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
result = uninstall_skill("sample-skill")
assert result["success"] is True
# Verify the skill was removed
skill_path = temp_skills_dir / "sample-skill"
assert not skill_path.exists()
def test_uninstall_nonexistent_skill(self, temp_skills_dir):
with mock.patch("EvoScientist.skills_manager.USER_SKILLS_DIR", temp_skills_dir):
result = uninstall_skill("nonexistent-skill")
assert result["success"] is False
assert "not found" in result["error"]
+17
View File
@@ -65,6 +65,23 @@ class TestFormatToolCompact:
result = format_tool_compact("edit_file", {"path": "f.py"})
assert result == "edit_file(f.py)"
# Memory file special display
def test_read_file_memory(self):
result = format_tool_compact("read_file", {"path": "/memory/MEMORY.md"})
assert result == "Reading memory"
def test_write_file_memory(self):
result = format_tool_compact("write_file", {"path": "/MEMORY.md"})
assert result == "Updating memory"
def test_write_file_memory_with_prefix(self):
result = format_tool_compact("write_file", {"path": "/memory/MEMORY.md"})
assert result == "Updating memory"
def test_edit_file_memory(self):
result = format_tool_compact("edit_file", {"path": "/memory/MEMORY.md"})
assert result == "Updating memory"
def test_glob(self):
result = format_tool_compact("glob", {"pattern": "*.py"})
assert result == "glob(*.py)"