diff --git a/README.md b/README.md index 8be56534..6a6267f8 100644 --- a/README.md +++ b/README.md @@ -5,18 +5,20 @@ **🐢✨The sassy AI code agent that makes IDEs look outdated** ✨🐢 [![Version](https://img.shields.io/pypi/v/code-puppy?style=for-the-badge&logo=python&label=Version&color=purple)](https://pypi.org/project/code-puppy/) -[![Downloads](https://img.shields.io/badge/Downloads-170k%2B-brightgreen?style=for-the-badge&logo=download)](https://pypi.org/project/code-puppy/) +[![Downloads](https://img.shields.io/badge/Downloads-100k%2B-brightgreen?style=for-the-badge&logo=download)](https://pypi.org/project/code-puppy/) [![Python](https://img.shields.io/badge/Python-3.11%2B-blue?style=for-the-badge&logo=python&logoColor=white)](https://python.org) [![License](https://img.shields.io/badge/License-MIT-green?style=for-the-badge)](LICENSE) [![Build Status](https://img.shields.io/badge/Build-Passing-brightgreen?style=for-the-badge&logo=github)](https://github.com/mpfaffenberger/code_puppy/actions) +[![Coverage](https://img.shields.io/badge/Coverage-95%25-brightgreen?style=for-the-badge)](https://github.com/mpfaffenberger/code_puppy) +[![Code Style](https://img.shields.io/badge/Code%20Style-Black-black?style=for-the-badge)](https://github.com/psf/black) [![Tests](https://img.shields.io/badge/Tests-Passing-success?style=for-the-badge&logo=pytest)](https://github.com/mpfaffenberger/code_puppy/tests) -[![OpenAI](https://img.shields.io/badge/OpenAI-GPT--5.2--Codex-orange?style=flat-square&logo=openai)](https://openai.com) +[![OpenAI](https://img.shields.io/badge/OpenAI-GPT--5-orange?style=flat-square&logo=openai)](https://openai.com) [![Gemini](https://img.shields.io/badge/Google-Gemini-blue?style=flat-square&logo=google)](https://ai.google.dev/) [![Anthropic](https://img.shields.io/badge/Anthropic-Claude-orange?style=flat-square&logo=anthropic)](https://anthropic.com) -[![Cerebras](https://img.shields.io/badge/Cerebras-GLM%204.7-red?style=flat-square)](https://cerebras.ai) -[![Z.AI](https://img.shields.io/badge/Z.AI-GLM%204.7-purple?style=flat-square)](https://z.ai/) -[![Synthetic](https://img.shields.io/badge/Synthetic-MINIMAX_M2.1-green?style=flat-square)](https://synthetic.new) +[![Cerebras](https://img.shields.io/badge/Cerebras-GLM%204.6-red?style=flat-square)](https://cerebras.ai) +[![Z.AI](https://img.shields.io/badge/Z.AI-GLM%204.6-purple?style=flat-square)](https://z.ai/) +[![Synthetic](https://img.shields.io/badge/Synthetic-MINIMAX_M2-green?style=flat-square)](https://synthetic.new) [![100% Open Source](https://img.shields.io/badge/100%25-Open%20Source-blue?style=for-the-badge)](https://github.com/mpfaffenberger/code_puppy) [![Pydantic AI](https://img.shields.io/badge/Pydantic-AI-success?style=for-the-badge)](https://github.com/pydantic/pydantic-ai) @@ -26,9 +28,6 @@ [![GitHub stars](https://img.shields.io/github/stars/mpfaffenberger/code_puppy?style=for-the-badge&logo=github)](https://github.com/mpfaffenberger/code_puppy/stargazers) [![GitHub forks](https://img.shields.io/github/forks/mpfaffenberger/code_puppy?style=for-the-badge&logo=github)](https://github.com/mpfaffenberger/code_puppy/network) -[![Discord](https://img.shields.io/badge/Discord-Community-purple?style=for-the-badge&logo=discord&logoColor=white)](https://discord.gg/SqYAaXVy) -[![Docs](https://img.shields.io/badge/Read-The%20Docs-blue?style=for-the-badge&logo=readthedocs)](https://code-puppy.dev) - **[⭐ Star this repo if you hate expensive IDEs! ⭐](#quick-start)** *"Who needs an IDE when you have 1024 angry puppies?"* - Someone, probably. @@ -62,32 +61,66 @@ uvx code-puppy -i ### UV (Recommended) -#### macOS / Linux - ```bash # Install UV if you don't have it curl -LsSf https://astral.sh/uv/install.sh | sh -uvx code-puppy +# Set UV to always use managed Python (one-time setup) +echo 'export UV_MANAGED_PYTHON=1' >> ~/.zshrc # or ~/.bashrc +source ~/.zshrc # or ~/.bashrc + +# Install and run code-puppy +uvx code-puppy -i +``` + +UV will automatically download the latest compatible Python version (3.11+) if your system doesn't have one. + +### pip (Alternative) + +```bash +pip install code-puppy ``` -#### Windows +*Note: pip installation requires your system Python to be 3.11 or newer.* + +### Permanent Python Management -On Windows, we recommend installing code-puppy as a global tool for the best experience with keyboard shortcuts (Ctrl+C/Ctrl+X cancellation): +To make UV always use managed Python versions (recommended): -```powershell -# Install UV if you don't have it (run in PowerShell as Admin) -powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex" +```bash +# Set environment variable permanently +echo 'export UV_MANAGED_PYTHON=1' >> ~/.zshrc # or ~/.bashrc +source ~/.zshrc # or ~/.bashrc -uvx code-puppy +# Now all UV commands will prefer managed Python installations +uvx code-puppy # No need for --managed-python flag anymore ``` -## Changelog (By Kittylog!) +### Verifying Python Version + +```bash +# Check which Python UV will use +uv python find -[πŸ“‹ View the full changelog on Kittylog](https://kittylog.app/c/mpfaffenberger/code_puppy) +# Or check the current project's Python +uv run python --version +``` ## Usage +### Custom Commands +Create markdown files in `.claude/commands/`, `.github/prompts/`, or `.agents/commands/` to define custom slash commands. The filename becomes the command name and the content runs as a prompt. + +```bash +# Create a custom command +echo "# Code Review + +Please review this code for security issues." > .claude/commands/review.md + +# Use it in Code Puppy +/review with focus on authentication +``` + ### Adding Models from models.dev πŸ†• While there are several models configured right out of the box from providers like Synthetic, Cerebras, OpenAI, Google, and Anthropic, Code Puppy integrates with [models.dev](https://models.dev) to let you browse and add models from **65+ providers** with a single command: @@ -155,18 +188,6 @@ The following environment variables control DBOS behavior: - `DBOS_SYSTEM_DATABASE_URL`: Database URL used by DBOS. Can point to a local SQLite file or a Postgres instance. Example: `postgresql://postgres:dbos@localhost:5432/postgres`. Default: `dbos_store.sqlite` file in the config directory. - `DBOS_APP_VERSION`: If set, Code Puppy uses it as the [DBOS application version](https://docs.dbos.dev/architecture#application-and-workflow-versions) and automatically tries to recover pending workflows for this version. Default: Code Puppy version + Unix timestamp in millisecond (disable automatic recovery). -### Custom Commands -Create markdown files in `.claude/commands/`, `.github/prompts/`, or `.agents/commands/` to define custom slash commands. The filename becomes the command name and the content runs as a prompt. - -```bash -# Create a custom command -echo "# Code Review - -Please review this code for security issues." > .claude/commands/review.md - -# Use it in Code Puppy -/review with focus on authentication -``` ## Requirements @@ -186,6 +207,58 @@ For examples and more information about agent rules, visit [https://agent.md](ht Use the `/mcp` command to manage MCP (list, start, stop, status, etc.) +Watch this video for examples! https://www.youtube.com/watch?v=1t1zEetOqlo + + +## Skills Plugin πŸ“š + +Code Puppy supports **Claude Code-compatible skills** - modular knowledge packages that extend the LLM with specialized expertise. + +### Quick Start + +```bash +/skill list # List installed skills +/skill add ~/my-skill # Install a skill +/skill info # View skill details +/skill show # Show full documentation +/skill remove # Uninstall skill +/skill refresh # Rescan skills directory +``` + +### Skills Directory + +``` +~/.code_puppy/skills/ +β”œβ”€β”€ pdf/ +β”‚ └── SKILL.md +└── docx/ + └── SKILL.md +``` + +### SKILL.md Format + +```yaml +--- +name: my-skill # kebab-case, required +description: "What it does" # required +license: "MIT" # optional +--- + +# Skill Documentation + +Full instructions go here... +``` + +### How It Works + +1. **Startup** β†’ Plugin scans `~/.code_puppy/skills/` for valid skills +2. **Prompts** β†’ Skill catalog is injected into system prompts +3. **Usage** β†’ LLM sees available skills and can request full docs + +See [Skills Plugin README](code_puppy/plugins/skills/README.md) for full documentation. + +--- + ## Round Robin Model Distribution Code Puppy supports **Round Robin model distribution** to help you overcome rate limits and distribute load across multiple AI models. This feature automatically cycles through configured models with each request, maximizing your API usage while staying within rate limits. diff --git a/code_puppy/agents/agent_code_puppy.py b/code_puppy/agents/agent_code_puppy.py index 2c105f2c..5fc15984 100644 --- a/code_puppy/agents/agent_code_puppy.py +++ b/code_puppy/agents/agent_code_puppy.py @@ -150,7 +150,7 @@ def get_system_prompt(self) -> str: Return your final response as a string output """ - prompt_additions = callbacks.on_load_prompt() - if len(prompt_additions): + prompt_additions = [p for p in callbacks.on_load_prompt() if p] + if prompt_additions: result += "\n".join(prompt_additions) return result diff --git a/code_puppy/agents/agent_planning.py b/code_puppy/agents/agent_planning.py index 3cc7f207..dfdefd4b 100644 --- a/code_puppy/agents/agent_planning.py +++ b/code_puppy/agents/agent_planning.py @@ -30,7 +30,6 @@ def get_available_tools(self) -> list[str]: "list_files", "read_file", "grep", - "agent_run_shell_command", "agent_share_your_reasoning", "list_agents", "invoke_agent", @@ -158,7 +157,7 @@ def get_system_prompt(self) -> str: IMPORTANT: Only when the user gives clear approval to proceed (such as "execute plan", "go ahead", "let's do it", "start", "begin", "proceed", "sounds good", or any equivalent phrase indicating they want to move forward), coordinate with the appropriate agents to implement your roadmap step by step, otherwise don't start invoking other tools such read file or other agents. """ - prompt_additions = callbacks.on_load_prompt() - if len(prompt_additions): + prompt_additions = [p for p in callbacks.on_load_prompt() if p] + if prompt_additions: result += "\n".join(prompt_additions) return result diff --git a/code_puppy/agents/prompt_reviewer.py b/code_puppy/agents/prompt_reviewer.py index b6d96326..d41fb453 100644 --- a/code_puppy/agents/prompt_reviewer.py +++ b/code_puppy/agents/prompt_reviewer.py @@ -139,7 +139,7 @@ def get_system_prompt(self) -> str: Remember: Great prompts lead to great results, but perfect is the enemy of good enough. """ - prompt_additions = callbacks.on_load_prompt() - if len(prompt_additions): + prompt_additions = [p for p in callbacks.on_load_prompt() if p] + if prompt_additions: result += "\n" + "\n".join(prompt_additions) return result diff --git a/code_puppy/plugins/skills/README.md b/code_puppy/plugins/skills/README.md new file mode 100644 index 00000000..a0ada430 --- /dev/null +++ b/code_puppy/plugins/skills/README.md @@ -0,0 +1,102 @@ +# Skills Plugin πŸ“š + +Claude Code-compatible skill support for Code Puppy. + +## What Are Skills? + +Skills are modular knowledge packages (`SKILL.md` files) that extend the LLM's capabilities with specialized expertise, workflows, and resources. + +## Quick Start + +```bash +# List installed skills +/skill list + +# Install a skill +/skill add ~/my-skills/pdf + +# View skill details +/skill info pdf + +# Show full documentation +/skill show pdf + +# Remove a skill +/skill remove pdf + +# Rescan skills directory +/skill refresh +``` + +## Skills Directory + +``` +~/.code_puppy/skills/ +β”œβ”€β”€ pdf/ +β”‚ β”œβ”€β”€ SKILL.md # Required: skill definition +β”‚ β”œβ”€β”€ scripts/ # Optional: helper scripts +β”‚ └── references/ # Optional: reference docs +└── docx/ + └── SKILL.md +``` + +## SKILL.md Format + +```yaml +--- +name: my-skill # kebab-case, required +description: "What it does" # required +license: "MIT" # optional +--- + +# Skill Title + +Full documentation goes here... +``` + +**Rules:** +- Must start with `---` (YAML frontmatter) +- `name` and `description` are required +- Body content follows the second `---` + +## How It Works + +1. **Startup** β†’ Plugin scans `~/.code_puppy/skills/` for valid skills +2. **Prompts** β†’ Skill catalog is injected into system prompts +3. **Usage** β†’ LLM sees available skills and can request full docs + +## Commands Reference + +| Command | Description | +|---------|-------------| +| `/skill` | Show help | +| `/skill list` | List installed skills | +| `/skill info ` | Show metadata | +| `/skill show ` | Show full SKILL.md | +| `/skill add ` | Install from directory | +| `/skill remove ` | Uninstall skill | +| `/skill refresh` | Rescan directory | + +## Creating a Skill + +```bash +mkdir -p ~/.code_puppy/skills/my-skill +cat > ~/.code_puppy/skills/my-skill/SKILL.md << 'EOF' +--- +name: my-skill +description: "My awesome skill for doing X" +--- + +# My Skill + +## When to Use +Use this skill when the user asks about X... + +## Workflow +1. Step one +2. Step two +3. Done! +EOF +``` + +Then run `/skill refresh` to load it. diff --git a/code_puppy/plugins/skills/__init__.py b/code_puppy/plugins/skills/__init__.py new file mode 100644 index 00000000..e68c0fd7 --- /dev/null +++ b/code_puppy/plugins/skills/__init__.py @@ -0,0 +1,7 @@ +"""Skills Plugin for Code Puppy - Claude Code compatible skill support.""" + +from code_puppy.plugins.skills.skill_loader import SkillLoader, SkillLoaderError +from code_puppy.plugins.skills.skill_manager import SkillManager +from code_puppy.plugins.skills.skill_types import SkillMetadata + +__all__ = ["SkillMetadata", "SkillLoader", "SkillLoaderError", "SkillManager"] diff --git a/code_puppy/plugins/skills/register_callbacks.py b/code_puppy/plugins/skills/register_callbacks.py new file mode 100644 index 00000000..edab5935 --- /dev/null +++ b/code_puppy/plugins/skills/register_callbacks.py @@ -0,0 +1,235 @@ +"""Skills Plugin for Code Puppy. + +Provides Claude Code-compatible skill support through the plugin system. +Registers callbacks for startup, commands, and prompt injection. +""" + +import logging +from pathlib import Path + +from code_puppy.callbacks import register_callback +from code_puppy.messaging import emit_info, emit_warning +from code_puppy.plugins.skills.skill_manager import SkillManager + +logger = logging.getLogger(__name__) + +# Global skill manager instance (lazy initialization) +_skill_manager: SkillManager | None = None + + +def _get_manager() -> SkillManager: + """Get or create the global SkillManager instance. + + Note: This uses simple lazy initialization. Thread-safety is not a concern + because Code Puppy's CLI is single-threaded and callbacks are invoked + sequentially from the main event loop. + """ + global _skill_manager + if _skill_manager is None: + _skill_manager = SkillManager() + return _skill_manager + + +# ================================================================ +# CALLBACK: startup +# ================================================================ +async def _on_startup() -> None: + """Initialize skill manager and show loaded count.""" + try: + manager = _get_manager() + count = manager.get_skill_count() + if count > 0: + emit_info(f"πŸ“š Skills plugin: Loaded {count} skill(s)") + except Exception as e: + logger.warning("Skills plugin failed to initialize: %s", e) + + +# ================================================================ +# CALLBACK: custom_command_help +# ================================================================ +def _skill_help() -> list[tuple[str, str]]: + """Return help entries for /skill commands.""" + return [ + ("skill", "Manage skills: /skill list|info|add|remove|refresh|show"), + ("skill list", "List all installed skills"), + ("skill info ", "Show skill details and metadata"), + ("skill add ", "Install a skill from directory"), + ("skill remove ", "Remove an installed skill"), + ("skill refresh", "Rescan the skills directory"), + ("skill show ", "Display the full SKILL.md content"), + ] + + +# ================================================================ +# CALLBACK: custom_command +# ================================================================ +def _handle_skill_command(command: str, name: str) -> str | None: + """Handle /skill commands. + + Args: + command: Full command string (e.g., "skill list") + name: Command name (e.g., "skill") + + Returns: + Command result if handled, None if not our command + """ + if name != "skill": + return None # Not our command + + parts = command.split() + subcommand = parts[1] if len(parts) > 1 else "help" + args = parts[2:] if len(parts) > 2 else [] + + manager = _get_manager() + + if subcommand == "list": + return _cmd_list(manager) + elif subcommand == "info" and args: + return _cmd_info(manager, args[0]) + elif subcommand == "add" and args: + return _cmd_add(manager, " ".join(args)) # Handle paths with spaces + elif subcommand == "remove" and args: + return _cmd_remove(manager, args[0]) + elif subcommand == "refresh": + return _cmd_refresh(manager) + elif subcommand == "show" and args: + return _cmd_show(manager, args[0]) + else: + return _cmd_help() + + +def _cmd_help() -> str: + """Show skill command help.""" + return """πŸ“š **Skills Plugin** + +Manage Claude Code-compatible skills for enhanced AI capabilities. + +**Commands:** + /skill list List all installed skills + /skill info Show skill details and metadata + /skill add Install a skill from directory + /skill remove Remove an installed skill + /skill refresh Rescan the skills directory + /skill show Display the full SKILL.md content + +**Skills Location:** ~/.code_puppy/skills/ +""" + + +def _cmd_list(manager: SkillManager) -> str: + """List installed skills.""" + skills = manager.list_skills() + + if not skills: + return ( + "πŸ“š **No skills installed**\n\nUse `/skill add ` to install a skill." + ) + + lines = [f"πŸ“š **Installed Skills ({len(skills)})**\n"] + for skill in skills: + desc = skill.description + if len(desc) > 60: + desc = desc[:57] + "..." + lines.append(f" **{skill.name}** - {desc}") + + lines.append("\nUse `/skill info ` for details.") + return "\n".join(lines) + + +def _cmd_info(manager: SkillManager, name: str) -> str: + """Show skill details.""" + info = manager.get_skill_info(name) + + if info is None: + emit_warning(f"Skill '{name}' not found") + return f"❌ Skill '{name}' not found" + + lines = [ + f"πŸ“š **Skill: {info['name']}**\n", + f"**Description:** {info['description']}", + f"**Path:** `{info['path']}`", + ] + if info.get("license"): + lines.append(f"**License:** {info['license']}") + + lines.append(f"\nUse `/skill show {name}` to see full documentation.") + return "\n".join(lines) + + +def _cmd_add(manager: SkillManager, path_str: str) -> str: + """Install a skill.""" + path = Path(path_str).expanduser() + success, message = manager.add_skill(path) + + if success: + emit_info(message) + return f"βœ… {message}" + else: + emit_warning(message) + return f"❌ {message}" + + +def _cmd_remove(manager: SkillManager, name: str) -> str: + """Remove a skill.""" + success, message = manager.remove_skill(name) + + if success: + emit_info(message) + return f"βœ… {message}" + else: + emit_warning(message) + return f"❌ {message}" + + +def _cmd_refresh(manager: SkillManager) -> str: + """Refresh skill catalog.""" + count = manager.refresh() + msg = f"πŸ“š Rescanned skills directory. Found {count} skill(s)." + emit_info(msg) + return msg + + +def _cmd_show(manager: SkillManager, name: str) -> str: + """Show full skill content.""" + body = manager.load_skill_body(name) + + if body is None: + emit_warning(f"Skill '{name}' not found") + return f"❌ Skill '{name}' not found" + + return f"πŸ“š **SKILL.md for {name}**\n\n{body}" + + +# ================================================================ +# CALLBACK: load_prompt +# ================================================================ +def _inject_skill_catalog() -> str | None: + """Inject skill catalog into system prompt.""" + manager = _get_manager() + catalog = manager.get_skill_catalog() + + if not catalog: + return None + + return f""" + +## Available Skills + +You have access to specialized skills. When a task matches a skill's +expertise, you can ask to see its detailed instructions. + +**Installed Skills:** +{catalog} + +To use a skill, mention "I'll use the [skill-name] skill for this task" +and request the full documentation if needed. +""" + + +# ================================================================ +# REGISTER ALL CALLBACKS +# ================================================================ +register_callback("startup", _on_startup) +register_callback("custom_command_help", _skill_help) +register_callback("custom_command", _handle_skill_command) +register_callback("load_prompt", _inject_skill_catalog) diff --git a/code_puppy/plugins/skills/skill_loader.py b/code_puppy/plugins/skills/skill_loader.py new file mode 100644 index 00000000..fade0c84 --- /dev/null +++ b/code_puppy/plugins/skills/skill_loader.py @@ -0,0 +1,168 @@ +"""Skill loader - parses SKILL.md files and extracts metadata.""" + +import logging +from pathlib import Path +from typing import Any + +import yaml + +from code_puppy.plugins.skills.skill_types import SkillMetadata + +logger = logging.getLogger(__name__) + + +class SkillLoaderError(Exception): + """Raised when a skill cannot be loaded.""" + + +class SkillLoader: + """Load and parse SKILL.md files. + + This loader handles Claude Code-compatible skill files with YAML frontmatter. + + Expected SKILL.md format: + --- + name: skill-name + description: What the skill does + license: Optional license + --- + + # Skill Body + + Instructions and documentation... + """ + + def __init__(self, skills_dir: Path) -> None: + """Initialize loader with base skills directory. + + Args: + skills_dir: Base directory where skills are stored. + Stored for future use by SkillManager. + """ + self.skills_dir = skills_dir + + def parse_frontmatter(self, skill_md: Path) -> SkillMetadata | None: + """Parse YAML frontmatter from SKILL.md file. + + Args: + skill_md: Path to the SKILL.md file + + Returns: + SkillMetadata if valid, None if parsing failed + """ + try: + content = skill_md.read_text(encoding="utf-8") + yaml_data, _ = self._extract_yaml_and_body(content) + + if yaml_data is None: + logger.warning("No YAML frontmatter found in %s", skill_md) + return None + + name = yaml_data.get("name") + description = yaml_data.get("description") + + if not name: + logger.warning("Missing 'name' field in %s", skill_md) + return None + if not description: + logger.warning("Missing 'description' field in %s", skill_md) + return None + + return SkillMetadata( + name=name, + description=description, + path=skill_md.parent, + license=yaml_data.get("license"), + ) + + except yaml.YAMLError as e: + logger.error("YAML parse error in %s: %s", skill_md, e) + return None + except OSError as e: + logger.error("Failed to read %s: %s", skill_md, e) + return None + except ValueError as e: + logger.warning("Invalid skill metadata in %s: %s", skill_md, e) + return None + + def load_body(self, skill_md: Path) -> str | None: + """Load the markdown body (excluding frontmatter). + + Args: + skill_md: Path to the SKILL.md file + + Returns: + Markdown body content, or None if file cannot be read + """ + try: + content = skill_md.read_text(encoding="utf-8") + _, body = self._extract_yaml_and_body(content) + return body + except OSError as e: + logger.error("Failed to read %s: %s", skill_md, e) + return None + + def get_resource(self, skill_path: Path, resource: str) -> str | None: + """Load a resource file from the skill directory. + + Args: + skill_path: Path to the skill directory + resource: Relative path to the resource file + + Returns: + Resource content, or None if not found + """ + resource_path = skill_path / resource + + # Security: Prevent path traversal using is_relative_to (Python 3.9+) + try: + resolved_resource = resource_path.resolve() + resolved_skill = skill_path.resolve() + # Check that resolved path is within skill directory + resolved_resource.relative_to(resolved_skill) + except (ValueError, OSError): + logger.warning("Path traversal attempt blocked: %s", resource) + return None + + try: + return resolved_resource.read_text(encoding="utf-8") + except OSError as e: + logger.warning("Failed to read resource %s: %s", resource, e) + return None + + def _extract_yaml_and_body(self, content: str) -> tuple[dict[str, Any] | None, str]: + """Split content into YAML dict and body string. + + Args: + content: Full SKILL.md content + + Returns: + Tuple of (yaml_dict or None, body_string) + """ + # Check for frontmatter markers + if not content.startswith("---"): + return None, content + + # Find the closing --- + lines = content.split("\n") + end_marker = -1 + for i, line in enumerate(lines[1:], start=1): + if line.strip() == "---": + end_marker = i + break + + if end_marker == -1: + # No closing marker found + return None, content + + # Extract YAML and body + yaml_content = "\n".join(lines[1:end_marker]) + body = "\n".join(lines[end_marker + 1 :]).strip() + + try: + yaml_data = yaml.safe_load(yaml_content) + if not isinstance(yaml_data, dict): + return None, content + return yaml_data, body + except yaml.YAMLError: + return None, content diff --git a/code_puppy/plugins/skills/skill_manager.py b/code_puppy/plugins/skills/skill_manager.py new file mode 100644 index 00000000..e055b82b --- /dev/null +++ b/code_puppy/plugins/skills/skill_manager.py @@ -0,0 +1,241 @@ +"""Skill manager - catalog management and skill operations.""" + +import logging +import shutil +from pathlib import Path +from typing import Any + +from code_puppy.plugins.skills.skill_loader import SkillLoader +from code_puppy.plugins.skills.skill_types import SkillMetadata + +logger = logging.getLogger(__name__) + +# Default skills directory +DEFAULT_SKILLS_DIR = Path.home() / ".code_puppy" / "skills" + +# Maximum description length for catalog display +MAX_DESCRIPTION_LENGTH = 80 + + +class SkillManager: + """Manage skill discovery, catalog, and operations. + + This manager handles: + - Scanning and loading skills from disk + - Providing a catalog for prompt injection + - CRUD operations for skill management + """ + + def __init__(self, skills_dir: Path | None = None) -> None: + """Initialize with skills directory. + + Args: + skills_dir: Directory where skills are stored. + Defaults to ~/.code_puppy/skills/ + """ + if skills_dir is None: + skills_dir = DEFAULT_SKILLS_DIR + self.skills_dir = skills_dir + self.loader = SkillLoader(skills_dir) + self._catalog: dict[str, SkillMetadata] = {} + self._refresh_catalog() + + def _refresh_catalog(self) -> None: + """Scan skills directory and rebuild catalog.""" + self._catalog.clear() + + if not self.skills_dir.exists(): + logger.debug("Skills directory does not exist: %s", self.skills_dir) + return + + if not self.skills_dir.is_dir(): + logger.warning("Skills path is not a directory: %s", self.skills_dir) + return + + for item in self.skills_dir.iterdir(): + if not item.is_dir(): + continue + if item.name.startswith(".") or item.name.startswith("_"): + continue + + skill_md = item / "SKILL.md" + if not skill_md.exists(): + logger.debug("No SKILL.md in %s, skipping", item.name) + continue + + metadata = self.loader.parse_frontmatter(skill_md) + if metadata is not None: + self._catalog[metadata.name] = metadata + logger.debug("Loaded skill: %s", metadata.name) + + def refresh(self) -> int: + """Rescan skills directory and rebuild catalog. + + Returns: + Number of skills found after refresh + """ + self._refresh_catalog() + return len(self._catalog) + + def get_skill_catalog(self) -> str: + """Generate catalog string for prompt injection. + + Returns: + Formatted string with skill names and descriptions, + or empty string if no skills are loaded. + """ + if not self._catalog: + return "" + + lines: list[str] = [] + for name in sorted(self._catalog.keys()): + skill = self._catalog[name] + description = skill.description + if len(description) > MAX_DESCRIPTION_LENGTH: + description = description[: MAX_DESCRIPTION_LENGTH - 3] + "..." + lines.append(f"- **{name}**: {description}") + + return "\n".join(lines) + + def get_skill_count(self) -> int: + """Return number of loaded skills.""" + return len(self._catalog) + + def get_skill(self, name: str) -> SkillMetadata | None: + """Get skill metadata by name. + + Args: + name: Skill identifier + + Returns: + SkillMetadata if found, None otherwise + """ + return self._catalog.get(name) + + def load_skill_body(self, skill_name: str) -> str | None: + """Load full SKILL.md body by name. + + Args: + skill_name: Skill identifier + + Returns: + Markdown body content, or None if skill not found + """ + skill = self.get_skill(skill_name) + if skill is None: + return None + + skill_md = skill.path / "SKILL.md" + return self.loader.load_body(skill_md) + + def get_resource(self, skill_name: str, resource_path: str) -> str | None: + """Load a resource from a skill. + + Args: + skill_name: Skill identifier + resource_path: Relative path to resource within skill directory + + Returns: + Resource content, or None if not found + """ + skill = self.get_skill(skill_name) + if skill is None: + return None + + return self.loader.get_resource(skill.path, resource_path) + + def list_skills(self) -> list[SkillMetadata]: + """Return list of all skills sorted by name.""" + return [self._catalog[name] for name in sorted(self._catalog.keys())] + + def add_skill(self, source_path: Path | str) -> tuple[bool, str]: + """Copy skill from source path to skills directory. + + Args: + source_path: Path to skill directory to install + + Returns: + Tuple of (success, message) + """ + source_path = Path(source_path) + + if not source_path.exists(): + return False, f"Source path does not exist: {source_path}" + + if not source_path.is_dir(): + return False, f"Source path is not a directory: {source_path}" + + skill_md = source_path / "SKILL.md" + if not skill_md.exists(): + return False, f"No SKILL.md found in {source_path}" + + # Parse to validate and get the name + metadata = self.loader.parse_frontmatter(skill_md) + if metadata is None: + return False, f"Invalid SKILL.md in {source_path}" + + # Create skills directory if needed + self.skills_dir.mkdir(parents=True, exist_ok=True) + + # Check if skill already exists + dest_path = self.skills_dir / metadata.name + if dest_path.exists(): + return False, f"Skill '{metadata.name}' already exists" + + try: + shutil.copytree(source_path, dest_path) + except OSError as e: + return False, f"Failed to copy skill: {e}" + + # Refresh to pick up new skill + self._refresh_catalog() + return True, f"Installed skill '{metadata.name}'" + + def remove_skill(self, skill_name: str) -> tuple[bool, str]: + """Remove a skill by name. + + Args: + skill_name: Skill identifier + + Returns: + Tuple of (success, message) + """ + skill = self.get_skill(skill_name) + if skill is None: + return False, f"Skill '{skill_name}' not found" + + # Safety: Verify path is within skills directory + try: + skill.path.resolve().relative_to(self.skills_dir.resolve()) + except ValueError: + logger.error("Skill path outside skills directory: %s", skill.path) + return False, f"Invalid skill path for '{skill_name}'" + + try: + shutil.rmtree(skill.path) + except OSError as e: + return False, f"Failed to remove skill: {e}" + + # Refresh catalog + self._refresh_catalog() + return True, f"Removed skill '{skill_name}'" + + def get_skill_info(self, skill_name: str) -> dict[str, Any] | None: + """Get detailed information about a skill. + + Args: + skill_name: Skill identifier + + Returns: + Dict with skill details, or None if not found + """ + skill = self.get_skill(skill_name) + if skill is None: + return None + + return { + "name": skill.name, + "description": skill.description, + "path": str(skill.path), + "license": skill.license, + } diff --git a/code_puppy/plugins/skills/skill_types.py b/code_puppy/plugins/skills/skill_types.py new file mode 100644 index 00000000..7057ee21 --- /dev/null +++ b/code_puppy/plugins/skills/skill_types.py @@ -0,0 +1,31 @@ +"""Data types for the Skills Plugin.""" + +from dataclasses import dataclass +from pathlib import Path + + +@dataclass +class SkillMetadata: + """Metadata extracted from a SKILL.md file's YAML frontmatter. + + Attributes: + name: Skill identifier (kebab-case, e.g., 'pdf-processor') + description: What the skill does and when to trigger it + path: Path to the skill directory containing SKILL.md + license: Optional license information + """ + + name: str + description: str + path: Path + license: str | None = None + + def __post_init__(self) -> None: + """Validate and normalize fields after initialization.""" + if not self.name or not self.name.strip(): + raise ValueError("Skill name cannot be empty") + if not self.description or not self.description.strip(): + raise ValueError("Skill description cannot be empty") + # Convert string path to Path object if needed + if isinstance(self.path, str): + self.path = Path(self.path) diff --git a/code_puppy/tools/agent_tools.py b/code_puppy/tools/agent_tools.py index 887a9bfb..ad3d7e52 100644 --- a/code_puppy/tools/agent_tools.py +++ b/code_puppy/tools/agent_tools.py @@ -7,7 +7,6 @@ import re import traceback from datetime import datetime -from functools import partial from pathlib import Path from typing import List, Set @@ -22,20 +21,18 @@ DATA_DIR, get_message_limit, get_use_dbos, - get_value, ) from code_puppy.messaging import ( SubAgentInvocationMessage, SubAgentResponseMessage, emit_error, emit_info, - emit_success, get_message_bus, get_session_context, set_session_context, ) +from code_puppy.model_factory import ModelFactory, make_model_settings from code_puppy.tools.common import generate_group_id -from code_puppy.tools.subagent_context import subagent_context # Set to track active subagent invocation tasks _active_subagent_tasks: Set[asyncio.Task] = set() @@ -249,12 +246,9 @@ def list_agents(context: RunContext) -> ListAgentsOutput: from rich.text import Text - from code_puppy.config import get_banner_color - - list_agents_color = get_banner_color("list_agents") emit_info( Text.from_markup( - f"\n[bold white on {list_agents_color}] LIST AGENTS [/bold white on {list_agents_color}]" + "\n[bold white on blue] LIST AGENTS [/bold white on blue]" ), message_group=group_id, ) @@ -415,9 +409,6 @@ async def invoke_agent( session_id = f"{session_id}-{hash_suffix}" # else: continuing existing session, use session_id as-is - # Lazy imports to avoid circular dependency - from code_puppy.agents.subagent_stream_handler import subagent_stream_handler - # Emit structured invocation message via MessageBus bus = get_message_bus() bus.emit( @@ -434,27 +425,7 @@ async def invoke_agent( previous_session_id = get_session_context() set_session_context(session_id) - # Set terminal session for browser-based terminal tools - # This uses contextvars which properly propagate through async tasks - from code_puppy.tools.browser.terminal_tools import ( - _terminal_session_var, - set_terminal_session, - ) - - terminal_session_token = set_terminal_session(f"terminal-{session_id}") - - # Set browser session for Camoufox browser tools (qa-kitten, etc.) - # This allows parallel agent invocations to each have their own browser - from code_puppy.tools.browser.camoufox_manager import ( - set_browser_session, - ) - - browser_session_token = set_browser_session(f"browser-{session_id}") - try: - # Lazy import to break circular dependency with messaging module - from code_puppy.model_factory import ModelFactory, make_model_settings - # Load the specified agent config agent_config = load_agent(agent_name) @@ -480,8 +451,8 @@ async def invoke_agent( from code_puppy import callbacks from code_puppy.model_utils import prepare_prompt_for_model - prompt_additions = callbacks.on_load_prompt() - if len(prompt_additions): + prompt_additions = [p for p in callbacks.on_load_prompt() if p] + if prompt_additions: instructions += "\n" + "\n".join(prompt_additions) # Handle claude-code models: swap instructions, and prepend system prompt only on first message @@ -497,118 +468,59 @@ async def invoke_agent( subagent_name = f"temp-invoke-agent-{session_id}" model_settings = make_model_settings(model_name) - # Get MCP servers for sub-agents (same as main agent) - from code_puppy.mcp_ import get_mcp_manager + temp_agent = Agent( + model=model, + instructions=instructions, + output_type=str, + retries=3, + history_processors=[agent_config.message_history_accumulator], + model_settings=model_settings, + ) + + # Register the tools that the agent needs + from code_puppy.tools import register_tools_for_agent - mcp_servers = [] - mcp_disabled = get_value("disable_mcp_servers") - if not ( - mcp_disabled and str(mcp_disabled).lower() in ("1", "true", "yes", "on") - ): - manager = get_mcp_manager() - mcp_servers = manager.get_servers_for_agent() + agent_tools = agent_config.get_available_tools() + register_tools_for_agent(temp_agent, agent_tools) if get_use_dbos(): from pydantic_ai.durable_exec.dbos import DBOSAgent - # For DBOS, create agent without MCP servers (to avoid serialization issues) - # and add them at runtime - temp_agent = Agent( - model=model, - instructions=instructions, - output_type=str, - retries=3, - toolsets=[], # MCP servers added separately for DBOS - history_processors=[agent_config.message_history_accumulator], - model_settings=model_settings, - ) - - # Register the tools that the agent needs - from code_puppy.tools import register_tools_for_agent - - agent_tools = agent_config.get_available_tools() - register_tools_for_agent(temp_agent, agent_tools) - - # Wrap with DBOS - no streaming for sub-agents - dbos_agent = DBOSAgent( - temp_agent, - name=subagent_name, - ) + dbos_agent = DBOSAgent(temp_agent, name=subagent_name) temp_agent = dbos_agent - # Store MCP servers to add at runtime - subagent_mcp_servers = mcp_servers - else: - # Non-DBOS path - include MCP servers directly in the agent - temp_agent = Agent( - model=model, - instructions=instructions, - output_type=str, - retries=3, - toolsets=mcp_servers, - history_processors=[agent_config.message_history_accumulator], - model_settings=model_settings, - ) - - # Register the tools that the agent needs - from code_puppy.tools import register_tools_for_agent - - agent_tools = agent_config.get_available_tools() - register_tools_for_agent(temp_agent, agent_tools) - - subagent_mcp_servers = None - # Run the temporary agent with the provided prompt as an asyncio task # Pass the message_history from the session to continue the conversation workflow_id = None # Track for potential cancellation - - # Always use subagent_stream_handler to silence output and update console manager - # This ensures all sub-agent output goes through the aggregated dashboard - stream_handler = partial(subagent_stream_handler, session_id=session_id) - - # Wrap the agent run in subagent context for tracking - with subagent_context(agent_name): - if get_use_dbos(): - # Generate a unique workflow ID for DBOS - ensures no collisions in back-to-back calls - workflow_id = _generate_dbos_workflow_id(group_id) - - # Add MCP servers to the DBOS agent's toolsets - # (temp_agent is discarded after this invocation, so no need to restore) - if subagent_mcp_servers: - temp_agent._toolsets = ( - temp_agent._toolsets + subagent_mcp_servers - ) - - with SetWorkflowID(workflow_id): - task = asyncio.create_task( - temp_agent.run( - prompt, - message_history=message_history, - usage_limits=UsageLimits( - request_limit=get_message_limit() - ), - event_stream_handler=stream_handler, - ) - ) - _active_subagent_tasks.add(task) - else: + if get_use_dbos(): + # Generate a unique workflow ID for DBOS - ensures no collisions in back-to-back calls + workflow_id = _generate_dbos_workflow_id(group_id) + with SetWorkflowID(workflow_id): task = asyncio.create_task( temp_agent.run( prompt, message_history=message_history, usage_limits=UsageLimits(request_limit=get_message_limit()), - event_stream_handler=stream_handler, ) ) _active_subagent_tasks.add(task) + else: + task = asyncio.create_task( + temp_agent.run( + prompt, + message_history=message_history, + usage_limits=UsageLimits(request_limit=get_message_limit()), + ) + ) + _active_subagent_tasks.add(task) - try: - result = await task - finally: - _active_subagent_tasks.discard(task) - if task.cancelled(): - if get_use_dbos() and workflow_id: - DBOS.cancel_workflow(workflow_id) + try: + result = await task + finally: + _active_subagent_tasks.discard(task) + if task.cancelled(): + if get_use_dbos() and workflow_id: + DBOS.cancel_workflow(workflow_id) # Extract the response from the result response = result.output @@ -635,23 +547,13 @@ async def invoke_agent( ) ) - # Emit clean completion summary - emit_success( - f"βœ“ {agent_name} completed successfully", message_group=group_id - ) - return AgentInvokeOutput( response=response, agent_name=agent_name, session_id=session_id ) - except Exception as e: - # Emit clean failure summary - emit_error(f"βœ— {agent_name} failed: {str(e)}", message_group=group_id) - - # Full traceback for debugging + except Exception: error_msg = f"Error invoking agent '{agent_name}': {traceback.format_exc()}" emit_error(error_msg, message_group=group_id) - return AgentInvokeOutput( response=None, agent_name=agent_name, @@ -662,13 +564,5 @@ async def invoke_agent( finally: # Restore the previous session context set_session_context(previous_session_id) - # Reset terminal session context - _terminal_session_var.reset(terminal_session_token) - # Reset browser session context - from code_puppy.tools.browser.camoufox_manager import ( - _browser_session_var, - ) - - _browser_session_var.reset(browser_session_token) return invoke_agent diff --git a/tests/plugins/__init__.py b/tests/plugins/__init__.py index da38704b..9ea65a92 100644 --- a/tests/plugins/__init__.py +++ b/tests/plugins/__init__.py @@ -1 +1 @@ -"""Test package for OAuth plugins.""" +"""Tests for Code Puppy plugins.""" diff --git a/tests/plugins/test_skill_loader.py b/tests/plugins/test_skill_loader.py new file mode 100644 index 00000000..88140125 --- /dev/null +++ b/tests/plugins/test_skill_loader.py @@ -0,0 +1,278 @@ +"""Tests for skill_loader module.""" + +from pathlib import Path +from textwrap import dedent + +import pytest + +from code_puppy.plugins.skills.skill_loader import SkillLoader + + +class TestSkillLoader: + """Tests for SkillLoader class.""" + + @pytest.fixture + def skills_dir(self, tmp_path: Path) -> Path: + """Create a temporary skills directory.""" + skills = tmp_path / "skills" + skills.mkdir() + return skills + + @pytest.fixture + def valid_skill(self, skills_dir: Path) -> Path: + """Create a valid skill with SKILL.md.""" + skill_dir = skills_dir / "test-skill" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text( + dedent(""" + --- + name: test-skill + description: A test skill for testing purposes + license: MIT + --- + + # Test Skill + + This is the body of the skill. + + ## Usage + + Instructions here. + """).strip(), + encoding="utf-8", + ) + return skill_dir + + @pytest.fixture + def skill_with_resources(self, skills_dir: Path) -> Path: + """Create a skill with resources directory.""" + skill_dir = skills_dir / "resource-skill" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text( + dedent(""" + --- + name: resource-skill + description: A skill with resources + --- + + # Resource Skill + """).strip(), + encoding="utf-8", + ) + + # Create resources + resources = skill_dir / "resources" + resources.mkdir() + (resources / "example.md").write_text("# Example Resource\n", encoding="utf-8") + return skill_dir + + @pytest.fixture + def loader(self, skills_dir: Path) -> SkillLoader: + """Create a SkillLoader instance.""" + return SkillLoader(skills_dir) + + def test_parse_valid_frontmatter( + self, loader: SkillLoader, valid_skill: Path + ) -> None: + """Test parsing a valid SKILL.md with name and description.""" + skill_md = valid_skill / "SKILL.md" + metadata = loader.parse_frontmatter(skill_md) + + assert metadata is not None + assert metadata.name == "test-skill" + assert metadata.description == "A test skill for testing purposes" + assert metadata.license == "MIT" + assert metadata.path == valid_skill + + def test_parse_missing_frontmatter( + self, loader: SkillLoader, skills_dir: Path + ) -> None: + """Test handling of file without --- markers.""" + skill_dir = skills_dir / "no-frontmatter" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text("# Just Markdown\n\nNo frontmatter here.", encoding="utf-8") + + metadata = loader.parse_frontmatter(skill_md) + assert metadata is None + + def test_parse_missing_name(self, loader: SkillLoader, skills_dir: Path) -> None: + """Test handling of SKILL.md without name field.""" + skill_dir = skills_dir / "no-name" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text( + dedent(""" + --- + description: Has description but no name + --- + + # Missing Name + """).strip(), + encoding="utf-8", + ) + + metadata = loader.parse_frontmatter(skill_md) + assert metadata is None + + def test_parse_missing_description( + self, loader: SkillLoader, skills_dir: Path + ) -> None: + """Test handling of SKILL.md without description field.""" + skill_dir = skills_dir / "no-desc" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text( + dedent(""" + --- + name: no-description-skill + --- + + # Missing Description + """).strip(), + encoding="utf-8", + ) + + metadata = loader.parse_frontmatter(skill_md) + assert metadata is None + + def test_load_body(self, loader: SkillLoader, valid_skill: Path) -> None: + """Test extracting markdown body after frontmatter.""" + skill_md = valid_skill / "SKILL.md" + body = loader.load_body(skill_md) + + assert body is not None + assert "# Test Skill" in body + assert "This is the body of the skill." in body + assert "---" not in body # Frontmatter should be excluded + assert "name:" not in body # YAML should be excluded + + def test_get_resource_exists( + self, loader: SkillLoader, skill_with_resources: Path + ) -> None: + """Test loading an existing resource file.""" + content = loader.get_resource(skill_with_resources, "resources/example.md") + + assert content is not None + assert "# Example Resource" in content + + def test_get_resource_not_found( + self, loader: SkillLoader, valid_skill: Path + ) -> None: + """Test handling of missing resource file.""" + content = loader.get_resource(valid_skill, "nonexistent.md") + assert content is None + + def test_get_resource_path_traversal_blocked( + self, loader: SkillLoader, valid_skill: Path + ) -> None: + """Test that path traversal attempts are blocked.""" + content = loader.get_resource(valid_skill, "../../../etc/passwd") + assert content is None + + def test_parse_invalid_yaml(self, loader: SkillLoader, skills_dir: Path) -> None: + """Test handling of invalid YAML in frontmatter.""" + skill_dir = skills_dir / "bad-yaml" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text( + dedent(""" + --- + name: [invalid yaml + description: unclosed bracket + --- + + # Bad YAML + """).strip(), + encoding="utf-8", + ) + + metadata = loader.parse_frontmatter(skill_md) + assert metadata is None + + def test_parse_nonexistent_file(self, loader: SkillLoader) -> None: + """Test handling of nonexistent file.""" + metadata = loader.parse_frontmatter(Path("/nonexistent/SKILL.md")) + assert metadata is None + + def test_empty_frontmatter(self, loader: SkillLoader, skills_dir: Path) -> None: + """Test handling of empty frontmatter markers.""" + skill_dir = skills_dir / "empty-fm" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text("---\n---\n# Body", encoding="utf-8") + + metadata = loader.parse_frontmatter(skill_md) + assert metadata is None + + def test_frontmatter_is_list_not_dict( + self, loader: SkillLoader, skills_dir: Path + ) -> None: + """Test handling of YAML frontmatter that's a list instead of dict.""" + skill_dir = skills_dir / "list-yaml" + skill_dir.mkdir() + skill_md = skill_dir / "SKILL.md" + skill_md.write_text( + dedent(""" + --- + - item1 + - item2 + --- + + # Body + """).strip(), + encoding="utf-8", + ) + + metadata = loader.parse_frontmatter(skill_md) + assert metadata is None + + def test_load_body_nonexistent_file(self, loader: SkillLoader) -> None: + """Test load_body with nonexistent file.""" + body = loader.load_body(Path("/nonexistent/SKILL.md")) + assert body is None + + +class TestExtractYamlAndBody: + """Tests for the _extract_yaml_and_body method.""" + + @pytest.fixture + def loader(self, tmp_path: Path) -> SkillLoader: + """Create a SkillLoader instance.""" + return SkillLoader(tmp_path) + + def test_no_frontmatter(self, loader: SkillLoader) -> None: + """Test content without frontmatter markers.""" + content = "# Just Markdown\n\nNo frontmatter." + yaml_data, body = loader._extract_yaml_and_body(content) + + assert yaml_data is None + assert body == content + + def test_unclosed_frontmatter(self, loader: SkillLoader) -> None: + """Test content with only opening --- marker.""" + content = "---\nname: test\n# No closing marker" + yaml_data, body = loader._extract_yaml_and_body(content) + + assert yaml_data is None + assert body == content + + def test_valid_frontmatter(self, loader: SkillLoader) -> None: + """Test valid frontmatter extraction.""" + content = dedent(""" + --- + name: my-skill + description: Test + --- + + # Body here + """).strip() + + yaml_data, body = loader._extract_yaml_and_body(content) + + assert yaml_data is not None + assert yaml_data["name"] == "my-skill" + assert yaml_data["description"] == "Test" + assert "# Body here" in body diff --git a/tests/plugins/test_skill_manager.py b/tests/plugins/test_skill_manager.py new file mode 100644 index 00000000..b640cf93 --- /dev/null +++ b/tests/plugins/test_skill_manager.py @@ -0,0 +1,332 @@ +"""Tests for skill_manager module.""" + +from pathlib import Path +from textwrap import dedent + +import pytest + +from code_puppy.plugins.skills.skill_manager import SkillManager + + +class TestSkillManager: + """Tests for SkillManager class.""" + + @pytest.fixture + def skills_dir(self, tmp_path: Path) -> Path: + """Create a temporary skills directory.""" + skills = tmp_path / "skills" + skills.mkdir() + return skills + + @pytest.fixture + def manager(self, skills_dir: Path) -> SkillManager: + """Create a SkillManager with empty skills directory.""" + return SkillManager(skills_dir) + + @pytest.fixture + def populated_skills_dir(self, skills_dir: Path) -> Path: + """Create skills directory with test skills.""" + # Skill 1: pdf-tools + pdf = skills_dir / "pdf-tools" + pdf.mkdir() + (pdf / "SKILL.md").write_text( + dedent(""" + --- + name: pdf-tools + description: PDF manipulation toolkit for extracting text and creating PDFs + license: MIT + --- + + # PDF Tools + + Instructions for PDF operations. + """).strip(), + encoding="utf-8", + ) + + # Skill 2: docx-helper + docx = skills_dir / "docx-helper" + docx.mkdir() + (docx / "SKILL.md").write_text( + dedent(""" + --- + name: docx-helper + description: Word document creation and editing + --- + + # DOCX Helper + + Instructions for Word docs. + """).strip(), + encoding="utf-8", + ) + + # Resources in pdf skill + resources = pdf / "resources" + resources.mkdir() + (resources / "example.md").write_text("# Example\n", encoding="utf-8") + + return skills_dir + + @pytest.fixture + def populated_manager(self, populated_skills_dir: Path) -> SkillManager: + """Create a SkillManager with test skills.""" + return SkillManager(populated_skills_dir) + + def test_manager_init_creates_catalog( + self, populated_manager: SkillManager + ) -> None: + """Test that manager scans skills on init.""" + assert populated_manager.get_skill_count() == 2 + assert populated_manager.get_skill("pdf-tools") is not None + assert populated_manager.get_skill("docx-helper") is not None + + def test_manager_empty_directory(self, manager: SkillManager) -> None: + """Test manager with no skills installed.""" + assert manager.get_skill_count() == 0 + assert manager.get_skill_catalog() == "" + assert manager.list_skills() == [] + + def test_manager_nonexistent_directory(self, tmp_path: Path) -> None: + """Test manager with nonexistent skills directory.""" + manager = SkillManager(tmp_path / "nonexistent") + assert manager.get_skill_count() == 0 + + def test_get_skill_catalog_format(self, populated_manager: SkillManager) -> None: + """Test catalog string format for prompt injection.""" + catalog = populated_manager.get_skill_catalog() + + # Should be sorted alphabetically + assert catalog.startswith("- **docx-helper**") + assert "- **pdf-tools**" in catalog + + # Should contain descriptions + assert "Word document" in catalog + assert "PDF manipulation" in catalog + + def test_get_skill_catalog_truncates_long_descriptions( + self, skills_dir: Path + ) -> None: + """Test that long descriptions are truncated.""" + long_desc = "A" * 100 # Longer than MAX_DESCRIPTION_LENGTH (80) + + skill = skills_dir / "long-desc" + skill.mkdir() + (skill / "SKILL.md").write_text( + dedent(f""" + --- + name: long-desc + description: {long_desc} + --- + + # Long Description Skill + """).strip(), + encoding="utf-8", + ) + + manager = SkillManager(skills_dir) + catalog = manager.get_skill_catalog() + + assert "..." in catalog + assert len(catalog.split("\n")[0]) < 100 + len("- **long-desc**: ") + + def test_get_skill_by_name(self, populated_manager: SkillManager) -> None: + """Test retrieving skill metadata by name.""" + skill = populated_manager.get_skill("pdf-tools") + + assert skill is not None + assert skill.name == "pdf-tools" + assert "PDF manipulation" in skill.description + assert skill.license == "MIT" + + def test_get_skill_not_found(self, populated_manager: SkillManager) -> None: + """Test handling of unknown skill name.""" + assert populated_manager.get_skill("nonexistent") is None + + def test_load_skill_body(self, populated_manager: SkillManager) -> None: + """Test loading full SKILL.md content.""" + body = populated_manager.load_skill_body("pdf-tools") + + assert body is not None + assert "# PDF Tools" in body + assert "Instructions for PDF operations" in body + + def test_load_skill_body_not_found(self, populated_manager: SkillManager) -> None: + """Test loading body for nonexistent skill.""" + assert populated_manager.load_skill_body("nonexistent") is None + + def test_get_resource(self, populated_manager: SkillManager) -> None: + """Test loading a resource from a skill.""" + content = populated_manager.get_resource("pdf-tools", "resources/example.md") + + assert content is not None + assert "# Example" in content + + def test_get_resource_skill_not_found( + self, populated_manager: SkillManager + ) -> None: + """Test getting resource from nonexistent skill.""" + assert populated_manager.get_resource("nonexistent", "file.md") is None + + def test_list_skills(self, populated_manager: SkillManager) -> None: + """Test listing all skills.""" + skills = populated_manager.list_skills() + + assert len(skills) == 2 + # Should be sorted alphabetically + assert skills[0].name == "docx-helper" + assert skills[1].name == "pdf-tools" + + def test_get_skill_info(self, populated_manager: SkillManager) -> None: + """Test getting detailed skill info.""" + info = populated_manager.get_skill_info("pdf-tools") + + assert info is not None + assert info["name"] == "pdf-tools" + assert ( + info["description"] + == "PDF manipulation toolkit for extracting text and creating PDFs" + ) + assert info["license"] == "MIT" + assert "path" in info + + def test_get_skill_info_not_found(self, populated_manager: SkillManager) -> None: + """Test getting info for nonexistent skill.""" + assert populated_manager.get_skill_info("nonexistent") is None + + def test_refresh_catalog( + self, populated_manager: SkillManager, populated_skills_dir: Path + ) -> None: + """Test rescanning skills directory.""" + # Add a new skill after manager creation + new_skill = populated_skills_dir / "new-skill" + new_skill.mkdir() + (new_skill / "SKILL.md").write_text( + dedent(""" + --- + name: new-skill + description: A newly added skill + --- + + # New Skill + """).strip(), + encoding="utf-8", + ) + + # Should not be visible yet + assert populated_manager.get_skill("new-skill") is None + + # Refresh and verify + count = populated_manager.refresh() + assert count == 3 + assert populated_manager.get_skill("new-skill") is not None + + +class TestSkillManagerAddRemove: + """Tests for add_skill and remove_skill operations.""" + + @pytest.fixture + def skills_dir(self, tmp_path: Path) -> Path: + """Create a temporary skills directory.""" + skills = tmp_path / "skills" + skills.mkdir() + return skills + + @pytest.fixture + def source_skill(self, tmp_path: Path) -> Path: + """Create a source skill to install.""" + source = tmp_path / "source-skill" + source.mkdir() + (source / "SKILL.md").write_text( + dedent(""" + --- + name: source-skill + description: A skill to be installed + --- + + # Source Skill + """).strip(), + encoding="utf-8", + ) + return source + + @pytest.fixture + def manager(self, skills_dir: Path) -> SkillManager: + """Create a SkillManager with empty skills directory.""" + return SkillManager(skills_dir) + + def test_add_skill_from_directory( + self, manager: SkillManager, source_skill: Path + ) -> None: + """Test installing skill from a path.""" + success, message = manager.add_skill(source_skill) + + assert success is True + assert "Installed skill 'source-skill'" in message + assert manager.get_skill("source-skill") is not None + assert manager.get_skill_count() == 1 + + def test_add_skill_nonexistent_source( + self, manager: SkillManager, tmp_path: Path + ) -> None: + """Test adding from nonexistent path.""" + success, message = manager.add_skill(tmp_path / "nonexistent") + + assert success is False + assert "does not exist" in message + + def test_add_skill_not_a_directory( + self, manager: SkillManager, tmp_path: Path + ) -> None: + """Test adding a file instead of directory.""" + file_path = tmp_path / "file.txt" + file_path.write_text("not a skill", encoding="utf-8") + + success, message = manager.add_skill(file_path) + + assert success is False + assert "not a directory" in message + + def test_add_skill_no_skill_md(self, manager: SkillManager, tmp_path: Path) -> None: + """Test adding directory without SKILL.md.""" + empty_dir = tmp_path / "empty" + empty_dir.mkdir() + + success, message = manager.add_skill(empty_dir) + + assert success is False + assert "No SKILL.md" in message + + def test_add_skill_already_exists( + self, manager: SkillManager, source_skill: Path + ) -> None: + """Test adding skill that already exists.""" + # Install first time + manager.add_skill(source_skill) + + # Try to install again + success, message = manager.add_skill(source_skill) + + assert success is False + assert "already exists" in message + + def test_remove_skill(self, manager: SkillManager, source_skill: Path) -> None: + """Test removing an installed skill.""" + # Install first + manager.add_skill(source_skill) + assert manager.get_skill_count() == 1 + + # Remove + success, message = manager.remove_skill("source-skill") + + assert success is True + assert "Removed skill" in message + assert manager.get_skill("source-skill") is None + assert manager.get_skill_count() == 0 + + def test_remove_skill_not_found(self, manager: SkillManager) -> None: + """Test removing nonexistent skill.""" + success, message = manager.remove_skill("nonexistent") + + assert success is False + assert "not found" in message diff --git a/tests/plugins/test_skill_plugin.py b/tests/plugins/test_skill_plugin.py new file mode 100644 index 00000000..5f8a6a27 --- /dev/null +++ b/tests/plugins/test_skill_plugin.py @@ -0,0 +1,378 @@ +"""Tests for the skills plugin registration and callbacks.""" + +from pathlib import Path +from textwrap import dedent +from typing import Any +from unittest.mock import patch + +import pytest + + +class TestPluginImports: + """Tests for plugin module imports.""" + + def test_plugin_imports(self) -> None: + """Test that plugin module can be imported.""" + # This will trigger callback registration + from code_puppy.plugins.skills import register_callbacks # noqa: F401 + + # If we get here, import succeeded + assert True + + def test_skill_manager_import(self) -> None: + """Test that SkillManager can be imported.""" + from code_puppy.plugins.skills.skill_manager import SkillManager + + assert SkillManager is not None + + +class TestStartupCallback: + """Tests for startup callback.""" + + @pytest.fixture + def skills_dir(self, tmp_path: Path) -> Path: + """Create a temporary skills directory.""" + skills = tmp_path / "skills" + skills.mkdir() + return skills + + @pytest.fixture + def populated_skills_dir(self, skills_dir: Path) -> Path: + """Create skills directory with a test skill.""" + skill = skills_dir / "test-skill" + skill.mkdir() + (skill / "SKILL.md").write_text( + dedent(""" + --- + name: test-skill + description: A test skill for testing + --- + # Test + """).strip(), + encoding="utf-8", + ) + return skills_dir + + @pytest.mark.asyncio + async def test_startup_with_skills(self, populated_skills_dir: Path) -> None: + """Test startup emits info when skills exist.""" + from code_puppy.plugins.skills.register_callbacks import _on_startup + from code_puppy.plugins.skills.skill_manager import SkillManager + + with ( + patch( + "code_puppy.plugins.skills.register_callbacks._skill_manager", + SkillManager(populated_skills_dir), + ), + patch( + "code_puppy.plugins.skills.register_callbacks.emit_info" + ) as mock_emit, + ): + await _on_startup() + + mock_emit.assert_called_once() + assert "1 skill" in mock_emit.call_args[0][0] + + @pytest.mark.asyncio + async def test_startup_no_skills(self, skills_dir: Path) -> None: + """Test startup is silent with no skills.""" + from code_puppy.plugins.skills.register_callbacks import _on_startup + from code_puppy.plugins.skills.skill_manager import SkillManager + + with ( + patch( + "code_puppy.plugins.skills.register_callbacks._skill_manager", + SkillManager(skills_dir), + ), + patch( + "code_puppy.plugins.skills.register_callbacks.emit_info" + ) as mock_emit, + ): + await _on_startup() + + mock_emit.assert_not_called() + + +class TestSkillHelp: + """Tests for the help callback.""" + + def test_skill_help_entries(self) -> None: + """Test help callback returns correct entries.""" + from code_puppy.plugins.skills.register_callbacks import _skill_help + + entries = _skill_help() + + assert isinstance(entries, list) + assert len(entries) >= 5 # At least 5 help entries + + # Check that main entries exist + names = [entry[0] for entry in entries] + assert "skill" in names + assert "skill list" in names + assert "skill info " in names + + +class TestSkillCommands: + """Tests for command handling.""" + + @pytest.fixture + def skills_dir(self, tmp_path: Path) -> Path: + """Create a temporary skills directory.""" + skills = tmp_path / "skills" + skills.mkdir() + return skills + + @pytest.fixture + def populated_skills_dir(self, skills_dir: Path) -> Path: + """Create skills directory with a test skill.""" + skill = skills_dir / "test-skill" + skill.mkdir() + (skill / "SKILL.md").write_text( + dedent(""" + --- + name: test-skill + description: A test skill for testing + license: MIT + --- + + # Test Skill + + This is the body. + """).strip(), + encoding="utf-8", + ) + return skills_dir + + @pytest.fixture + def mock_manager(self, populated_skills_dir: Path) -> Any: + """Create a mock SkillManager using temp directory.""" + from code_puppy.plugins.skills.skill_manager import SkillManager + + return SkillManager(populated_skills_dir) + + def test_handle_skill_list(self, mock_manager: Any) -> None: + """Test /skill list command.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_list + + result = _cmd_list(mock_manager) + + assert "Installed Skills" in result + assert "test-skill" in result + + def test_handle_skill_info(self, mock_manager: Any) -> None: + """Test /skill info command.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_info + + result = _cmd_info(mock_manager, "test-skill") + + assert "test-skill" in result + assert "Description" in result + assert "A test skill for testing" in result + + def test_handle_skill_info_not_found(self, mock_manager: Any) -> None: + """Test /skill info for nonexistent skill.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_info + + result = _cmd_info(mock_manager, "nonexistent") + + assert "not found" in result + + def test_handle_skill_show(self, mock_manager: Any) -> None: + """Test /skill show command.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_show + + result = _cmd_show(mock_manager, "test-skill") + + assert "SKILL.md" in result + assert "# Test Skill" in result + assert "This is the body" in result + + def test_handle_skill_show_not_found(self, mock_manager: Any) -> None: + """Test /skill show for nonexistent skill.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_show + + result = _cmd_show(mock_manager, "nonexistent") + + assert "not found" in result + + def test_handle_skill_refresh(self, mock_manager: Any) -> None: + """Test /skill refresh command.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_refresh + + result = _cmd_refresh(mock_manager) + + assert "Rescanned" in result + assert "1 skill" in result + + def test_handle_skill_help(self) -> None: + """Test /skill (no subcommand) shows help.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_help + + result = _cmd_help() + + assert "Skills Plugin" in result + assert "/skill list" in result + assert "/skill add" in result + + def test_handle_unknown_command_returns_none(self) -> None: + """Test handling of non-skill commands (returns None).""" + from code_puppy.plugins.skills.register_callbacks import _handle_skill_command + + # Should return None for non-skill commands + result = _handle_skill_command("other command", "other") + assert result is None + + def test_handle_skill_command_routes_to_list(self, mock_manager: Any) -> None: + """Test /skill list routes correctly.""" + from code_puppy.plugins.skills.register_callbacks import _handle_skill_command + + with patch( + "code_puppy.plugins.skills.register_callbacks._get_manager", + return_value=mock_manager, + ): + result = _handle_skill_command("skill list", "skill") + + assert result is not None + assert "Installed Skills" in result or "No skills" in result + + def test_handle_skill_command_default_to_help(self) -> None: + """Test /skill with no subcommand shows help.""" + from code_puppy.plugins.skills.register_callbacks import _handle_skill_command + + result = _handle_skill_command("skill", "skill") + + assert result is not None + assert "Skills Plugin" in result + + +class TestPromptInjection: + """Tests for prompt injection callback.""" + + @pytest.fixture + def skills_dir(self, tmp_path: Path) -> Path: + """Create a temporary skills directory.""" + skills = tmp_path / "skills" + skills.mkdir() + return skills + + @pytest.fixture + def populated_skills_dir(self, skills_dir: Path) -> Path: + """Create skills directory with test skills.""" + skill = skills_dir / "test-skill" + skill.mkdir() + (skill / "SKILL.md").write_text( + dedent(""" + --- + name: test-skill + description: A test skill + --- + # Test + """).strip(), + encoding="utf-8", + ) + return skills_dir + + def test_prompt_injection_with_skills(self, populated_skills_dir: Path) -> None: + """Test catalog is injected when skills exist.""" + from code_puppy.plugins.skills.register_callbacks import _inject_skill_catalog + from code_puppy.plugins.skills.skill_manager import SkillManager + + # Patch the global manager + with patch( + "code_puppy.plugins.skills.register_callbacks._skill_manager", + SkillManager(populated_skills_dir), + ): + result = _inject_skill_catalog() + + assert result is not None + assert "Available Skills" in result + assert "test-skill" in result + + def test_prompt_injection_no_skills(self, skills_dir: Path) -> None: + """Test no injection when no skills.""" + from code_puppy.plugins.skills.register_callbacks import _inject_skill_catalog + from code_puppy.plugins.skills.skill_manager import SkillManager + + # Patch with empty skills directory + with patch( + "code_puppy.plugins.skills.register_callbacks._skill_manager", + SkillManager(skills_dir), + ): + result = _inject_skill_catalog() + + assert result is None + + +class TestAddRemoveCommands: + """Tests for add and remove command handlers.""" + + @pytest.fixture + def skills_dir(self, tmp_path: Path) -> Path: + """Create a temporary skills directory.""" + skills = tmp_path / "skills" + skills.mkdir() + return skills + + @pytest.fixture + def source_skill(self, tmp_path: Path) -> Path: + """Create a source skill to install.""" + source = tmp_path / "source-skill" + source.mkdir() + (source / "SKILL.md").write_text( + dedent(""" + --- + name: source-skill + description: A skill to install + --- + # Source + """).strip(), + encoding="utf-8", + ) + return source + + @pytest.fixture + def manager(self, skills_dir: Path) -> Any: + """Create a SkillManager with empty skills directory.""" + from code_puppy.plugins.skills.skill_manager import SkillManager + + return SkillManager(skills_dir) + + def test_cmd_add_success(self, manager: Any, source_skill: Path) -> None: + """Test successful skill installation.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_add + + result = _cmd_add(manager, str(source_skill)) + + assert "βœ…" in result + assert "Installed" in result + + def test_cmd_add_not_found(self, manager: Any, tmp_path: Path) -> None: + """Test add with nonexistent path.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_add + + result = _cmd_add(manager, str(tmp_path / "nonexistent")) + + assert "❌" in result + assert "does not exist" in result + + def test_cmd_remove_success(self, manager: Any, source_skill: Path) -> None: + """Test successful skill removal.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_add, _cmd_remove + + # Install first + _cmd_add(manager, str(source_skill)) + + # Remove + result = _cmd_remove(manager, "source-skill") + + assert "βœ…" in result + assert "Removed" in result + + def test_cmd_remove_not_found(self, manager: Any) -> None: + """Test remove nonexistent skill.""" + from code_puppy.plugins.skills.register_callbacks import _cmd_remove + + result = _cmd_remove(manager, "nonexistent") + + assert "❌" in result + assert "not found" in result diff --git a/tests/plugins/test_skill_types.py b/tests/plugins/test_skill_types.py new file mode 100644 index 00000000..ee04bf5d --- /dev/null +++ b/tests/plugins/test_skill_types.py @@ -0,0 +1,80 @@ +"""Tests for skill_types module.""" + +from pathlib import Path + +import pytest + +from code_puppy.plugins.skills.skill_types import SkillMetadata + + +class TestSkillMetadata: + """Tests for SkillMetadata dataclass.""" + + def test_skill_metadata_creation(self) -> None: + """Test creating a SkillMetadata with required fields.""" + metadata = SkillMetadata( + name="test-skill", + description="A test skill for testing", + path=Path("/tmp/skills/test-skill"), + ) + assert metadata.name == "test-skill" + assert metadata.description == "A test skill for testing" + assert metadata.path == Path("/tmp/skills/test-skill") + assert metadata.license is None + + def test_skill_metadata_with_license(self) -> None: + """Test creating a SkillMetadata with optional license field.""" + metadata = SkillMetadata( + name="licensed-skill", + description="A skill with a license", + path=Path("/tmp/skills/licensed"), + license="MIT", + ) + assert metadata.license == "MIT" + + def test_skill_metadata_path_type(self) -> None: + """Test that path is always a Path object.""" + # String path should be converted to Path + metadata = SkillMetadata( + name="path-test", + description="Testing path conversion", + path="/tmp/skills/path-test", # type: ignore[arg-type] + ) + assert isinstance(metadata.path, Path) + assert metadata.path == Path("/tmp/skills/path-test") + + def test_skill_metadata_empty_name_raises(self) -> None: + """Test that empty name raises ValueError.""" + with pytest.raises(ValueError, match="name cannot be empty"): + SkillMetadata( + name="", + description="Valid description", + path=Path("/tmp"), + ) + + def test_skill_metadata_empty_description_raises(self) -> None: + """Test that empty description raises ValueError.""" + with pytest.raises(ValueError, match="description cannot be empty"): + SkillMetadata( + name="valid-name", + description="", + path=Path("/tmp"), + ) + + def test_skill_metadata_whitespace_name_raises(self) -> None: + """Test that whitespace-only name raises ValueError.""" + with pytest.raises(ValueError, match="name cannot be empty"): + SkillMetadata( + name=" ", + description="Valid description", + path=Path("/tmp"), + ) + + def test_skill_metadata_whitespace_description_raises(self) -> None: + """Test that whitespace-only description raises ValueError.""" + with pytest.raises(ValueError, match="description cannot be empty"): + SkillMetadata( + name="valid-name", + description=" \t\n ", + path=Path("/tmp"), + ) diff --git a/tests/test_agent_tools.py b/tests/test_agent_tools.py index 67cc7f9d..47466150 100644 --- a/tests/test_agent_tools.py +++ b/tests/test_agent_tools.py @@ -58,7 +58,7 @@ def test_invoke_agent_includes_prompt_additions(self): ) # Get prompt additions to verify they exist - prompt_additions = callbacks.on_load_prompt() + prompt_additions = [p for p in callbacks.on_load_prompt() if p] # Verify we have file permission prompt additions assert len(prompt_additions) > 0 diff --git a/tests/tools/test_command_runner_core.py b/tests/tools/test_command_runner_core.py index 8b786258..01955ad2 100644 --- a/tests/tools/test_command_runner_core.py +++ b/tests/tools/test_command_runner_core.py @@ -254,14 +254,12 @@ def test_kill_all_concurrent_access_thread_safety(self, monkeypatch): mock_kill = MagicMock() monkeypatch.setattr(command_runner_module, "_kill_process_group", mock_kill) - # Create multiple fake processes with unique PIDs for this test + # Create multiple fake processes processes = [] - test_pids = set() for i in range(5): proc = MagicMock() proc.poll.return_value = None - proc.pid = 9000 + i # Use high PIDs unlikely to conflict - test_pids.add(proc.pid) + proc.pid = 1000 + i processes.append(proc) _register_process(proc) @@ -286,20 +284,18 @@ def kill_worker(): for thread in threads: thread.join() - # Verify that our test processes were handled - # (mock_kill may be called for processes from other tests too) - killed_pids = { - call.args[0].pid for call in mock_kill.call_args_list if call.args - } - our_killed_pids = killed_pids & test_pids + # With concurrent access, multiple threads may each see and attempt to kill + # the same processes before they're unregistered, so call_count can exceed + # the number of processes (up to num_threads * num_processes in worst case) + num_threads = 3 + assert mock_kill.call_count <= len(processes) * num_threads - # At least some of our test processes should have been killed - # (due to thread races, not all threads may see all processes) - assert len(our_killed_pids) > 0 or mock_kill.call_count > 0 + # Registry should be empty + verify_processes_registered = len(list(_RUNNING_PROCESSES)) + assert verify_processes_registered == 0 - # Our test processes should no longer be in the registry - remaining_test_procs = [p for p in _RUNNING_PROCESSES if p.pid in test_pids] - assert len(remaining_test_procs) == 0 + # At least one thread should have successfully killed processes + assert any(r > 0 for r in results) or mock_kill.call_count > 0 def test_kill_all_tracks_killed_processes(self, monkeypatch): """Test that killed PIDs are added to _USER_KILLED_PROCESSES."""