diff --git a/.gitignore b/.gitignore index 66afd97..d32b41d 100644 --- a/.gitignore +++ b/.gitignore @@ -31,7 +31,7 @@ workspace/ skills/ memory/ .deno_cache/ -*test.ipynb +*.ipynb *CLAUDE.md *AGENTS.md *meals/ \ No newline at end of file diff --git a/EvoScientist/cli.py b/EvoScientist/cli.py index 25507d8..74c6881 100644 --- a/EvoScientist/cli.py +++ b/EvoScientist/cli.py @@ -26,9 +26,12 @@ from typing import Any, Optional import typer # type: ignore[import-untyped] from prompt_toolkit import PromptSession # type: ignore[import-untyped] +from prompt_toolkit.completion import Completer, Completion # type: ignore[import-untyped] from prompt_toolkit.history import FileHistory # type: ignore[import-untyped] from prompt_toolkit.auto_suggest import AutoSuggestFromHistory # type: ignore[import-untyped] from prompt_toolkit.formatted_text import HTML # type: ignore[import-untyped] +from prompt_toolkit.shortcuts import CompleteStyle # type: ignore[import-untyped] +from prompt_toolkit.styles import Style as PtStyle # type: ignore[import-untyped] from rich.panel import Panel # type: ignore[import-untyped] from rich.text import Text # type: ignore[import-untyped] from rich.table import Table # type: ignore[import-untyped] @@ -200,17 +203,8 @@ def print_banner( for line, color in zip(EVOSCIENTIST_ASCII_LINES, _GRADIENT_COLORS): console.print(Text(line, style=f"{color} bold")) info = Text() - info.append(" Thread: ", style="dim") - info.append(thread_id, style="yellow") - if workspace_dir: - info.append("\n Workspace: ", style="dim") - 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(_shorten_path(trimmed), style="cyan") if model or provider or mode: - info.append("\n ", style="dim") + info.append(" ", style="dim") parts = [] if model: parts.append(("Model: ", model)) @@ -223,20 +217,9 @@ def print_banner( info.append(" ", style="dim") info.append(label, style="dim") info.append(value, style="magenta") - info.append("\n Commands: ", style="dim") - info.append("/exit", style="bold") - info.append(", ", style="dim") - info.append("/new", style="bold") - info.append(", ", style="dim") - info.append("/thread", style="bold") - 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") - info.append(", ", style="dim") - info.append("/mcp", style="bold") + info.append("\n Type ", style="#ffe082") + info.append("/", style="#ffe082 bold") + info.append(" for commands", style="#ffe082") console.print(info) console.print() @@ -254,13 +237,13 @@ def _cmd_list_skills() -> None: skills = list_skills(include_system=False) if not skills: - console.print("[dim]No user skills installed.[/dim]") + console.print("[dim]No user-installed skills.[/dim]") console.print("[dim]Install with:[/dim] /install-skill ") 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)}):") + console.print(f"[bold]User-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]") @@ -620,6 +603,45 @@ def _auto_start_channel(agent: Any, thread_id: str, allowed_senders_csv: str) -> _print_channel_panel([("iMessage", False, str(e))]) +_SLASH_COMMANDS = [ + ("/thread", "Show thread ID, workspace & memory dir"), + ("/new", "Start a new session"), + ("/skills", "List installed skills"), + ("/install-skill", "Add a skill from path or GitHub"), + ("/uninstall-skill", "Remove an installed skill"), + ("/mcp", "Manage MCP servers"), + ("/channel", "Configure messaging channels"), + ("/exit", "Quit EvoScientist"), +] + +_COMPLETION_STYLE = PtStyle.from_dict({ + "completion-menu": "bg:default noreverse nounderline noitalic", + "completion-menu.completion": "bg:default #888888 noreverse", + "completion-menu.completion.current": "bg:default default bold noreverse", + "completion-menu.meta.completion": "bg:default #888888 noreverse", + "completion-menu.meta.completion.current": "bg:default default bold noreverse", + "scrollbar.background": "bg:default", + "scrollbar.button": "bg:default", +}) + + +class SlashCommandCompleter(Completer): + """Autocomplete for slash commands — triggers when input starts with '/'.""" + + def get_completions(self, document, complete_event): + text = document.text_before_cursor + if not text.startswith("/"): + return + for cmd, desc in _SLASH_COMMANDS: + if cmd.startswith(text): + yield Completion( + cmd, + start_position=-len(text), + display=f"{cmd:<40}", + display_meta=desc, + ) + + def cmd_interactive( agent: Any, show_thinking: bool = True, @@ -656,7 +678,10 @@ def cmd_interactive( session = PromptSession( history=FileHistory(history_file), auto_suggest=AutoSuggestFromHistory(), - enable_history_search=True, + completer=SlashCommandCompleter(), + complete_style=CompleteStyle.COLUMN, + complete_while_typing=True, + style=_COMPLETION_STYLE, ) def _print_separator(): @@ -785,7 +810,7 @@ def cmd_interactive( while state["running"]: try: user_input = await session.prompt_async( - HTML('> ') + HTML('❯ ') ) user_input = user_input.strip() diff --git a/EvoScientist/skills/find-skills/scripts/install_skill.py b/EvoScientist/skills/find-skills/scripts/install_skill.py index f89194a..5bfbb30 100644 --- a/EvoScientist/skills/find-skills/scripts/install_skill.py +++ b/EvoScientist/skills/find-skills/scripts/install_skill.py @@ -1,16 +1,18 @@ #!/usr/bin/env python3 """Install a skill from GitHub into a local skills directory. -Self-contained installer — no external dependencies beyond git. +Thin CLI wrapper around EvoScientist.skills_manager — all clone, resolve, +and validation logic lives in the core module. Usage examples: # Install from a GitHub URL (auto-detects repo, ref, path) python install_skill.py --url https://github.com/anthropics/skills/tree/main/excel - # Install from repo + path - python install_skill.py --repo anthropics/skills --path excel + # Install from shorthand + python install_skill.py --url anthropics/skills@excel - # Install multiple skills from the same repo + # Install from repo + path(s) + python install_skill.py --repo anthropics/skills --path excel python install_skill.py --repo anthropics/skills --path excel --path pdf # Install with a specific git ref @@ -20,131 +22,20 @@ Usage examples: from __future__ import annotations import argparse -import os -import re -import shutil -import subprocess import sys -import tempfile + +from EvoScientist.skills_manager import install_skill, _parse_github_url -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"] +def _build_source(repo: str, ref: str | None, path: str | None) -> str: + """Build a GitHub URL string that install_skill() can consume.""" + if ref and path: + return f"https://github.com/{repo}/tree/{ref}/{path}" 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 copy_skill(src: str, dest_dir: str) -> str: - """Copy a skill directory to the destination. - - Returns: - The skill name (directory basename). - """ - skill_name = os.path.basename(src.rstrip("/")) - target = os.path.join(dest_dir, skill_name) - - if os.path.exists(target): - shutil.rmtree(target) - print(f" Replaced existing: {skill_name}") - - shutil.copytree(src, target) - return skill_name - - -def validate_skill(path: str) -> bool: - """Check that a directory looks like a valid skill (has SKILL.md).""" - return os.path.isfile(os.path.join(path, "SKILL.md")) - - -def install( - repo: str, - paths: list[str], - ref: str | None, - dest: str, -) -> list[str]: - """Install skill(s) from a GitHub repo. - - Returns: - List of installed skill names. - """ - os.makedirs(dest, exist_ok=True) - installed: list[str] = [] - - with tempfile.TemporaryDirectory(prefix="skill-install-") as tmp: - clone_dir = os.path.join(tmp, "repo") - print(f"Cloning {repo}" + (f" @{ref}" if ref else "") + "...") - clone_repo(repo, ref, clone_dir) - - if not paths: - # No path specified — treat entire repo as a single skill - if validate_skill(clone_dir): - name = copy_skill(clone_dir, dest) - installed.append(name) - else: - # List top-level directories that look like skills - for entry in sorted(os.listdir(clone_dir)): - entry_path = os.path.join(clone_dir, entry) - if os.path.isdir(entry_path) and validate_skill(entry_path): - name = copy_skill(entry_path, dest) - installed.append(name) - - if not installed: - print("No valid skills found in repository root.", file=sys.stderr) - else: - for p in paths: - skill_path = os.path.join(clone_dir, p.strip("/")) - if not os.path.isdir(skill_path): - print(f" Path not found: {p}", file=sys.stderr) - continue - if not validate_skill(skill_path): - print(f" No SKILL.md in: {p}", file=sys.stderr) - continue - name = copy_skill(skill_path, dest) - installed.append(name) - - return installed + return f"https://github.com/{repo}/tree/{ref}" + if path: + return f"{repo}@{path}" + return f"https://github.com/{repo}" def main() -> int: @@ -178,11 +69,10 @@ def main() -> int: ) args = parser.parse_args() - dest = args.dest # Parse source if args.url: - repo, ref, path = parse_github_url(args.url) + repo, ref, path = _parse_github_url(args.url) ref = args.ref or ref paths = [path] if path else args.path else: @@ -190,22 +80,28 @@ def main() -> int: ref = args.ref paths = args.path - try: - installed = install(repo, paths, ref, dest) - except RuntimeError as e: - print(f"Error: {e}", file=sys.stderr) - return 1 + # Install each path (or the whole repo if no paths given) + sources = [_build_source(repo, ref, p) for p in paths] if paths else [_build_source(repo, ref, None)] + + installed = [] + for source in sources: + print(f"Installing from: {source}") + result = install_skill(source, dest_dir=args.dest) + if result["success"]: + print(f" Installed: {result['name']} ({result.get('description', '')})") + installed.append(result["name"]) + else: + print(f" Failed: {result['error']}", file=sys.stderr) if installed: - print(f"\nInstalled {len(installed)} skill(s) to {dest}/:") + print(f"\nInstalled {len(installed)} skill(s) to {args.dest}/:") for name in installed: print(f" - {name}") + return 0 else: print("No skills were installed.", file=sys.stderr) return 1 - return 0 - if __name__ == "__main__": raise SystemExit(main()) diff --git a/EvoScientist/skills_manager.py b/EvoScientist/skills_manager.py index 394e462..5e54973 100644 --- a/EvoScientist/skills_manager.py +++ b/EvoScientist/skills_manager.py @@ -77,6 +77,11 @@ def _parse_skill_md(skill_md_path: Path) -> dict[str, str]: try: frontmatter = yaml.safe_load(frontmatter_match.group(1)) + if not isinstance(frontmatter, dict): + return { + "name": skill_md_path.parent.name, + "description": "(empty frontmatter)", + } return { "name": frontmatter.get("name", skill_md_path.parent.name), "description": frontmatter.get("description", "(no description)"), @@ -123,6 +128,9 @@ def _parse_github_url(url: str) -> tuple[str, str | None, str | None]: raise ValueError(f"Cannot parse GitHub URL: {url}") +_CLONE_TIMEOUT = 120 # seconds + + def _clone_repo(repo: str, ref: str | None, dest: str) -> None: """Shallow-clone a GitHub repo.""" clone_url = f"https://github.com/{repo}.git" @@ -131,7 +139,10 @@ def _clone_repo(repo: str, ref: str | None, dest: str) -> None: cmd += ["--branch", ref] cmd += [clone_url, dest] - result = subprocess.run(cmd, capture_output=True, text=True) + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=_CLONE_TIMEOUT) + except subprocess.TimeoutExpired: + raise RuntimeError(f"git clone timed out after {_CLONE_TIMEOUT}s for {repo}") if result.returncode != 0: raise RuntimeError(f"git clone failed: {result.stderr.strip()}") @@ -159,6 +170,42 @@ def _validate_skill_dir(path: Path) -> bool: return (path / "SKILL.md").is_file() +def _find_skill_in_tree(root: str, skill_name: str) -> Path | None: + """Walk a directory tree to find a subdirectory named *skill_name* containing SKILL.md. + + Skips hidden directories (starting with '.'). + + Returns: + The absolute Path to the skill directory, or None. + """ + for dirpath, dirnames, _files in os.walk(root): + # Prune hidden directories + dirnames[:] = [d for d in dirnames if not d.startswith(".")] + if os.path.basename(dirpath) == skill_name: + candidate = Path(dirpath) + if _validate_skill_dir(candidate): + return candidate + return None + + +# Allowed pattern for skill names: alphanumeric, hyphens, underscores +_VALID_SKILL_NAME = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*$") + + +def _sanitize_name(name: str) -> str | None: + """Validate and sanitize a skill name. + + Returns the cleaned name, or None if invalid. + """ + name = name.strip() + if not name or not _VALID_SKILL_NAME.match(name): + return None + # Block path traversal components + if ".." in name or "/" in name or "\\" in name: + return None + return name + + def install_skill(source: str, dest_dir: str | None = None) -> dict: """Install a skill from a local path or GitHub URL. @@ -197,10 +244,14 @@ def _install_from_local(source: str, dest_dir: str) -> dict: # Parse SKILL.md to get the skill name skill_info = _parse_skill_md(source_path / "SKILL.md") - skill_name = skill_info["name"] + skill_name = _sanitize_name(skill_info["name"]) + if not skill_name: + return {"success": False, "error": f"Invalid skill name in SKILL.md: {skill_info['name']!r}"} - # Destination path - target_path = Path(dest_dir) / skill_name + # Destination path — resolve and verify it stays inside dest_dir + target_path = (Path(dest_dir) / skill_name).resolve() + if not str(target_path).startswith(str(Path(dest_dir).resolve())): + return {"success": False, "error": f"Skill name escapes destination: {skill_info['name']!r}"} # Remove existing if present if target_path.exists(): @@ -238,34 +289,48 @@ def _install_from_github(source: str, dest_dir: str) -> dict: 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 + # Validate — if the direct path doesn't have SKILL.md, try auto-resolve + if not skill_source.exists() or not _validate_skill_dir(skill_source): + if path: + # The shorthand path (e.g. "canvas-design") may be nested deeper + # Walk the tree to find a directory with that name + SKILL.md + skill_name_hint = path.rstrip("/").rsplit("/", 1)[-1] + resolved = _find_skill_in_tree(clone_dir, skill_name_hint) + if resolved: + skill_source = resolved + else: + return {"success": False, "error": f"No SKILL.md found at '{path}' (also searched subdirectories) in: {source}"} + else: + # No path specified — list available 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: + if len(found_skills) == 1: + # Only one skill in repo — just install it + skill_source = Path(clone_dir) / found_skills[0] + elif found_skills: return { "success": False, "error": ( f"Multiple skills found in repo. " - f"Please specify one: {', '.join(found_skills)}" + f"Please specify one: {', '.join(sorted(found_skills))}" ), } - - return {"success": False, "error": f"No SKILL.md found in: {source}"} + else: + 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 + skill_name = _sanitize_name(skill_info["name"]) + if not skill_name: + return {"success": False, "error": f"Invalid skill name in SKILL.md: {skill_info['name']!r}"} + + target_path = (Path(dest_dir) / skill_name).resolve() + if not str(target_path).startswith(str(Path(dest_dir).resolve())): + return {"success": False, "error": f"Skill name escapes destination: {skill_info['name']!r}"} if target_path.exists(): shutil.rmtree(target_path) @@ -348,8 +413,14 @@ def uninstall_skill(name: str) -> dict: - success: bool - error: error message (if failed) """ - user_dir = Path(USER_SKILLS_DIR) - target_path = user_dir / name + user_dir = Path(USER_SKILLS_DIR).resolve() + + # Validate name to prevent path traversal + clean_name = _sanitize_name(name) + if not clean_name: + return {"success": False, "error": f"Invalid skill name: {name!r}"} + + target_path = (user_dir / clean_name).resolve() if not target_path.exists(): # Try to find by directory name (in case name differs from dir name) @@ -358,15 +429,15 @@ def uninstall_skill(name: str) -> dict: 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 + if info["name"] == clean_name: + found = entry.resolve() 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) + # Check resolved path is still inside user_dir if not str(target_path).startswith(str(user_dir)): return {"success": False, "error": f"Cannot uninstall system skill: {name}"} diff --git a/EvoScientist/tools.py b/EvoScientist/tools.py index 1abb61b..651b0c0 100644 --- a/EvoScientist/tools.py +++ b/EvoScientist/tools.py @@ -122,35 +122,51 @@ async def tavily_search( @tool(parse_docstring=True) def skill_manager( - action: Literal["install", "list", "uninstall"], + action: Literal["install", "list", "uninstall", "info"], source: str = "", name: str = "", + include_system: bool = False, ) -> str: - """Manage user skills: install, list, or uninstall. + """Manage user-installable skills: install from GitHub or local path, list available skills, get details, 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) + Actions and required parameters: - 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" + action="install" (requires source): + Install a skill. The source can be: + - GitHub shorthand: "owner/repo@skill-name" (e.g. "anthropics/skills@peft") + - GitHub URL: "https://github.com/owner/repo/tree/main/skill-name" + - Local path: "./my-skill" or "/path/to/skill" + Nested skills are auto-resolved — if the skill is not at the repo root, subdirectories are searched automatically. + + action="list": + List installed skills. By default only shows user-installed skills. + Set include_system=True to also show built-in system skills (peft, accelerate, flash-attention, etc.). + + action="info" (requires name): + Get details (description, source, path) about a specific skill by name. + Searches both user and system skills. + + action="uninstall" (requires name): + Remove a user-installed skill by name. System skills cannot be uninstalled. Args: - action: One of "install", "list", or "uninstall" - source: For install - local path or GitHub URL/shorthand - name: For uninstall - skill name to remove + action: The operation to perform — "install", "list", "info", or "uninstall" + source: Required for install — GitHub shorthand, GitHub URL, or local directory path + name: Required for info and uninstall — the skill name (e.g. "peft", "my-custom-skill") + include_system: Only for list — set True to include built-in system skills in the output Returns: Result message """ - from .skills_manager import install_skill, list_skills, uninstall_skill + from .skills_manager import install_skill, list_skills, uninstall_skill, get_skill_info if action == "install": if not source: - return "Error: 'source' is required for install action" + return ( + "Error: 'source' is required for install action. " + "Provide a GitHub shorthand (e.g. source='owner/repo@skill-name'), " + "a GitHub URL, or a local directory path." + ) result = install_skill(source) if result["success"]: return ( @@ -163,25 +179,59 @@ def skill_manager( return f"Failed to install skill: {result['error']}" elif action == "list": - skills = list_skills(include_system=False) + skills = list_skills(include_system=include_system) 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}") + if include_system: + return "No skills found." + return "No user skills installed. Use action='install' to add skills, or set include_system=True to see built-in skills." + user_skills = [s for s in skills if s.source == "user"] + system_skills = [s for s in skills if s.source == "system"] + lines = [] + if user_skills: + lines.append(f"User Skills ({len(user_skills)}):") + for skill in user_skills: + lines.append(f" - {skill.name}: {skill.description}") + if system_skills: + if lines: + lines.append("") + lines.append(f"System Skills ({len(system_skills)}):") + for skill in system_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" + return ( + "Error: 'name' is required for uninstall action. " + "Use action='list' first to see installed skill names." + ) result = uninstall_skill(name) if result["success"]: return f"Successfully uninstalled skill: {name}" else: return f"Failed to uninstall skill: {result['error']}" + elif action == "info": + if not name: + return ( + "Error: 'name' is required for info action. " + "Use action='list' with include_system=True to see all available skill names." + ) + info = get_skill_info(name) + if not info: + return ( + f"Skill not found: {name}. " + f"Use action='list' with include_system=True to see all available skills." + ) + return ( + f"Name: {info.name}\n" + f"Description: {info.description}\n" + f"Source: {info.source}\n" + f"Path: {info.path}" + ) + else: - return f"Unknown action: {action}. Use 'install', 'list', or 'uninstall'." + return f"Unknown action: {action}. Use 'install', 'list', 'uninstall', or 'info'." @tool(parse_docstring=True)