diff --git a/src/cmdforge/runner.py b/src/cmdforge/runner.py index b68a83d..ca209fe 100644 --- a/src/cmdforge/runner.py +++ b/src/cmdforge/runner.py @@ -533,15 +533,22 @@ def execute_prompt_step( prompt = substitute_variables(prompt_template, variables, warn_non_scalar=verbose) - # Inject profile system prompt if specified + # Determine provider + provider = provider_override or substitute_variables(step.provider, variables, warn_non_scalar=verbose) + + # Build context from the inside out so the final order is: + # profile system prompt -> skills -> user prompt. + if step.skills is not None: + from .skills import inject_skills + skill_names = None if "*" in step.skills else step.skills + prompt = inject_skills(prompt, provider, skill_names) + if step.profile: profile = load_profile(step.profile) if profile and profile.system_prompt: - # Prepend system prompt to user prompt prompt = f"{profile.system_prompt}\n\n---\n\n{prompt}" - # Determine provider - provider = provider_override or substitute_variables(step.provider, variables, warn_non_scalar=verbose) + # Structured or plain-text # Plain text mode - bypass structured output if step.plain_text: diff --git a/src/cmdforge/skills.py b/src/cmdforge/skills.py new file mode 100644 index 0000000..84d7ad8 --- /dev/null +++ b/src/cmdforge/skills.py @@ -0,0 +1,221 @@ +"""Provider-attached skills in the Agent Skills open standard. + +Each provider can have skills at:: + + ~/.cmdforge/providers//skills//SKILL.md + +Every ``SKILL.md`` must contain the required Agent Skills frontmatter. The +directory name is the canonical skill identifier and must match ``name``. +""" + +import re +import sys +from dataclasses import dataclass, field +from pathlib import Path +from typing import List, Optional + +import yaml + +PROVIDERS_DIR = Path.home() / ".cmdforge" / "providers" + +_PROVIDER_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]*$") +_SKILL_NAME_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +_MISSING_POLICIES = {"warn", "error", "ignore"} + + +@dataclass(frozen=True) +class Skill: + """A single loaded skill.""" + + name: str + directory: str + description: str + content: str = "" + + def __str__(self) -> str: + return self.content + + +@dataclass +class SkillSet: + """Collection of skills for one provider, deterministically ordered.""" + + provider_name: str + skills: List[Skill] = field(default_factory=list) + + def get( + self, + requested: Optional[List[str]] = None, + on_missing: str = "warn", + ) -> List[Skill]: + """Return enabled skills, reporting missing ones.""" + if on_missing not in _MISSING_POLICIES: + raise ValueError( + f"Invalid missing-skill policy '{on_missing}'; expected one of " + f"{sorted(_MISSING_POLICIES)}" + ) + + available = {skill.name: skill for skill in self.skills} + if requested is None: + return list(self.skills) + + enabled: List[Skill] = [] + for name in requested: + if name in available: + enabled.append(available[name]) + continue + + message = ( + f"Skill '{name}' requested for provider '{self.provider_name}' " + "but not found in skills directory" + ) + if on_missing == "error": + raise KeyError(message) + if on_missing == "warn": + print(f"[skills] {message}", file=sys.stderr) + return enabled + + +def _validate_provider_name(name: str) -> None: + """Validate a provider name before using it as a path component.""" + if not isinstance(name, str) or not _PROVIDER_NAME_RE.fullmatch(name): + raise ValueError( + f"Invalid provider name '{name}'. Must match {_PROVIDER_NAME_RE.pattern}" + ) + + +def _validate_skill_name(name: str, provider: str = "") -> None: + """Validate the Agent Skills canonical name rules.""" + context = f" under provider '{provider}'" if provider else "" + if ( + not isinstance(name, str) + or not 1 <= len(name) <= 64 + or not _SKILL_NAME_RE.fullmatch(name) + ): + raise ValueError( + f"Invalid skill name '{name}'{context}. Skill names must be 1-64 " + "lowercase letters, numbers, or single hyphens, and may not start " + "or end with a hyphen" + ) + + +def _parse_frontmatter(text: str) -> tuple[dict, str]: + """Parse the mandatory YAML frontmatter from a ``SKILL.md`` file.""" + lines = text.splitlines(keepends=True) + if not lines or lines[0].rstrip("\r\n") != "---": + raise ValueError("SKILL.md must begin with YAML frontmatter") + + closing_index = next( + (index for index, line in enumerate(lines[1:], start=1) + if line.rstrip("\r\n") == "---"), + None, + ) + if closing_index is None: + raise ValueError("SKILL.md frontmatter is not terminated") + + frontmatter_raw = "".join(lines[1:closing_index]).strip() + if not frontmatter_raw: + raise ValueError("SKILL.md frontmatter is empty; metadata is required") + try: + metadata = yaml.safe_load(frontmatter_raw) + except yaml.YAMLError as exc: + raise ValueError(f"Invalid YAML in SKILL.md frontmatter: {exc}") from exc + if not isinstance(metadata, dict): + raise ValueError("SKILL.md frontmatter must be a YAML mapping") + + content = "".join(lines[closing_index + 1:]).lstrip("\r\n").rstrip() + return metadata, content + + +def _validate_metadata(metadata: dict, directory_name: str, provider: str) -> tuple[str, str]: + """Validate required Agent Skills metadata and return normalized values.""" + if "name" not in metadata: + raise ValueError(f"SKILL.md for '{directory_name}' is missing required 'name'") + name = metadata["name"] + _validate_skill_name(name, provider) + if name != directory_name: + raise ValueError( + f"SKILL.md frontmatter name '{name}' does not match skill directory " + f"name '{directory_name}' under provider '{provider}'" + ) + + if "description" not in metadata: + raise ValueError( + f"SKILL.md for '{directory_name}' is missing required 'description'" + ) + description = metadata["description"] + if not isinstance(description, str) or not description.strip(): + raise ValueError( + f"SKILL.md description for '{directory_name}' must be a non-empty string" + ) + if len(description) > 1024: + raise ValueError( + f"SKILL.md description for '{directory_name}' must not exceed 1024 characters" + ) + return name, description.strip() + + +def _reject_symlink(path: Path, label: str) -> None: + if path.is_symlink(): + raise ValueError(f"Refusing symlinked {label}: {path}") + + +def load_provider_skills(provider_name: str) -> SkillSet: + """Load and validate all skills configured for a provider.""" + _validate_provider_name(provider_name) + + provider_dir = PROVIDERS_DIR / provider_name + skills_dir = provider_dir / "skills" + _reject_symlink(provider_dir, "provider directory") + _reject_symlink(skills_dir, "skills directory") + if not skills_dir.is_dir(): + return SkillSet(provider_name=provider_name) + + skills: List[Skill] = [] + for entry in sorted(skills_dir.iterdir(), key=lambda path: path.name): + _reject_symlink(entry, "skill directory") + if not entry.is_dir(): + continue + _validate_skill_name(entry.name, provider_name) + + skill_md = entry / "SKILL.md" + _reject_symlink(skill_md, "SKILL.md") + if not skill_md.is_file(): + continue + try: + text = skill_md.read_text(encoding="utf-8") + except OSError as exc: + raise OSError(f"Cannot read {skill_md}: {exc}") from exc + + metadata, content = _parse_frontmatter(text) + name, description = _validate_metadata( + metadata, entry.name, provider_name + ) + skills.append( + Skill( + name=name, + directory=str(entry), + description=description, + content=content, + ) + ) + + return SkillSet(provider_name=provider_name, skills=skills) + + +def inject_skills( + prompt: str, + provider_name: str, + skill_names: Optional[List[str]] = None, + on_missing: str = "warn", +) -> str: + """Prepend enabled provider skills to a prompt.""" + skillset = load_provider_skills(provider_name) + enabled = skillset.get(skill_names, on_missing=on_missing) + if not enabled: + return prompt + + skill_context = "\n\n".join( + f"## Skill: {skill.name}\n\n{skill.content}" for skill in enabled + ) + return f"{skill_context}\n\n---\n\n{prompt}" diff --git a/src/cmdforge/tool.py b/src/cmdforge/tool.py index 6b29169..459fba8 100644 --- a/src/cmdforge/tool.py +++ b/src/cmdforge/tool.py @@ -130,6 +130,24 @@ class PromptStep: max_retries: int = 1 # Retry count on validation failure plain_text: bool = False # Bypass structured output enforcement max_tokens: Optional[int] = None # Max output tokens (provider-dependent) + skills: Optional[List[str]] = None # Skill names to enable for this step + + def __post_init__(self) -> None: + if self.skills is None: + return + if not isinstance(self.skills, list): + raise ValueError("skills must be a list of skill names") + if any(not isinstance(name, str) for name in self.skills): + raise ValueError("skills entries must be strings") + if "*" in self.skills and self.skills != ["*"]: + raise ValueError("'*' must be the only entry when enabling all skills") + if len(self.skills) != len(set(self.skills)): + raise ValueError("skills must not contain duplicate names") + + from .skills import _validate_skill_name + for name in self.skills: + if name != "*": + _validate_skill_name(name) def to_dict(self) -> dict: d = { @@ -154,6 +172,8 @@ class PromptStep: d["plain_text"] = self.plain_text if self.max_tokens: d["max_tokens"] = self.max_tokens + if self.skills is not None: + d["skills"] = self.skills return d @classmethod @@ -178,7 +198,8 @@ class PromptStep: output_schema=data.get("output_schema"), max_retries=data.get("max_retries", 1), plain_text=data.get("plain_text", False), - max_tokens=max_tokens + max_tokens=max_tokens, + skills=data.get("skills"), ) diff --git a/tests/test_skills.py b/tests/test_skills.py new file mode 100644 index 0000000..3d975ef --- /dev/null +++ b/tests/test_skills.py @@ -0,0 +1,438 @@ +"""Tests for provider-attached skills system.""" + +import textwrap +from types import SimpleNamespace + +import pytest + +from cmdforge.skills import ( + Skill, + SkillSet, + _parse_frontmatter, + _validate_provider_name, + _validate_skill_name, + load_provider_skills, + inject_skills, +) +from cmdforge.providers import ProviderResult +from cmdforge.runner import execute_prompt_step +from cmdforge.tool import PromptStep + + +def write_skill(root, provider, name, content="Skill instructions.", description="A useful skill"): + directory = root / provider / "skills" / name + directory.mkdir(parents=True) + (directory / "SKILL.md").write_text( + f"---\nname: {name}\ndescription: {description}\n---\n{content}", + encoding="utf-8", + ) + return directory + + +class TestValidation: + def test_valid_provider_name(self): + _validate_provider_name("claude") + _validate_provider_name("openrouter") + _validate_provider_name("my-provider_1") + + def test_invalid_provider_name_slash(self): + with pytest.raises(ValueError): + _validate_provider_name("claude/evil") + + def test_invalid_provider_name_dots(self): + with pytest.raises(ValueError): + _validate_provider_name("../../etc") + + def test_invalid_provider_name_empty(self): + with pytest.raises(ValueError): + _validate_provider_name("") + + def test_valid_skill_name(self): + _validate_skill_name("python-utils", "claude") + + @pytest.mark.parametrize("name", ["Python", "python_utils", "-python", "python-", "python--utils"]) + def test_invalid_standard_skill_names(self, name): + with pytest.raises(ValueError, match="Invalid skill name"): + _validate_skill_name(name, "claude") + + def test_skill_name_max_length(self): + _validate_skill_name("a" * 64, "claude") + with pytest.raises(ValueError, match="1-64"): + _validate_skill_name("a" * 65, "claude") + + def test_invalid_skill_name_slash(self): + with pytest.raises(ValueError): + _validate_skill_name("a/b", "claude") + + def test_invalid_skill_name_dotdot(self): + with pytest.raises(ValueError): + _validate_skill_name("..", "claude") + + +class TestFrontmatter: + def test_basic(self): + text = textwrap.dedent("""\ + --- + name: python-utils + description: Python helpers + --- + Use list comprehensions. + """) + meta, content = _parse_frontmatter(text) + assert meta["name"] == "python-utils" + assert meta["description"] == "Python helpers" + assert content == "Use list comprehensions." + + def test_no_frontmatter(self): + with pytest.raises(ValueError, match="must begin"): + _parse_frontmatter("Just raw content") + + def test_unterminated_frontmatter(self): + with pytest.raises(ValueError, match="not terminated"): + _parse_frontmatter("---\nname: test\ndescription: Test") + + def test_invalid_yaml_frontmatter(self): + text = "---\n: invalid yaml\n---\ncontent" + with pytest.raises(ValueError, match="Invalid YAML"): + _parse_frontmatter(text) + + def test_non_dict_frontmatter(self): + text = "---\n- list item\n---\ncontent" + with pytest.raises(ValueError, match="mapping"): + _parse_frontmatter(text) + + +class TestLoadProviderSkills: + @pytest.fixture + def temp_providers_dir(self, tmp_path, monkeypatch): + monkeypatch.setattr("cmdforge.skills.PROVIDERS_DIR", tmp_path) + return tmp_path + + def test_no_directory(self, temp_providers_dir): + skills = load_provider_skills("nonexistent") + assert isinstance(skills, SkillSet) + assert skills.skills == [] + + def test_empty_skills_dir(self, temp_providers_dir): + d = temp_providers_dir / "claude" / "skills" + d.mkdir(parents=True) + skills = load_provider_skills("claude") + assert skills.skills == [] + + def test_loads_skill(self, temp_providers_dir): + d = temp_providers_dir / "claude" / "skills" / "python" + d.mkdir(parents=True) + (d / "SKILL.md").write_text(textwrap.dedent("""\ + --- + name: python + description: Python coding patterns + --- + Use type hints always. + """)) + + skillset = load_provider_skills("claude") + assert len(skillset.skills) == 1 + skill = skillset.skills[0] + assert skill.name == "python" + assert skill.description == "Python coding patterns" + assert "type hints" in skill.content + + def test_skips_non_md_files(self, temp_providers_dir): + d = write_skill(temp_providers_dir, "claude", "python") + (d / "readme.txt").write_text("ignore me") + + skillset = load_provider_skills("claude") + assert len(skillset.skills) == 1 + + def test_sorted_by_name(self, temp_providers_dir): + for name in ("z-skill", "a-skill", "m-skill"): + write_skill(temp_providers_dir, "claude", name) + + skillset = load_provider_skills("claude") + names = [s.name for s in skillset.skills] + assert names == ["a-skill", "m-skill", "z-skill"] + + def test_frontmatter_name_must_match_directory(self, temp_providers_dir): + d = temp_providers_dir / "claude" / "skills" / "python" + d.mkdir(parents=True) + (d / "SKILL.md").write_text( + "---\nname: wrong-name\ndescription: Wrong\n---\ncontent" + ) + + with pytest.raises(ValueError, match="does not match"): + load_provider_skills("claude") + + def test_invalid_provider_name_traversal(self, temp_providers_dir): + with pytest.raises(ValueError): + load_provider_skills("../secrets") + + def test_skips_non_directories(self, temp_providers_dir): + d = temp_providers_dir / "claude" / "skills" + d.mkdir(parents=True) + (d / "file.txt").write_text("not a skill dir") + skillset = load_provider_skills("claude") + assert skillset.skills == [] + + def test_no_skill_md(self, temp_providers_dir): + d = temp_providers_dir / "claude" / "skills" / "empty-dir" + d.mkdir(parents=True) + skillset = load_provider_skills("claude") + assert skillset.skills == [] + + @pytest.mark.parametrize( + ("frontmatter", "message"), + [ + ("description: Useful", "required 'name'"), + ("name: python", "required 'description'"), + ("name: python\ndescription: ''", "non-empty"), + ], + ) + def test_required_metadata(self, temp_providers_dir, frontmatter, message): + directory = temp_providers_dir / "claude" / "skills" / "python" + directory.mkdir(parents=True) + (directory / "SKILL.md").write_text( + f"---\n{frontmatter}\n---\ncontent" + ) + with pytest.raises(ValueError, match=message): + load_provider_skills("claude") + + def test_description_max_length(self, temp_providers_dir): + write_skill( + temp_providers_dir, + "claude", + "python", + description="x" * 1025, + ) + with pytest.raises(ValueError, match="1024"): + load_provider_skills("claude") + + def test_rejects_symlinked_provider_directory(self, temp_providers_dir, tmp_path): + outside = tmp_path / "outside-provider" + outside.mkdir() + (temp_providers_dir / "claude").symlink_to(outside, target_is_directory=True) + with pytest.raises(ValueError, match="symlinked provider"): + load_provider_skills("claude") + + def test_rejects_symlinked_skill_directory(self, temp_providers_dir, tmp_path): + skills_dir = temp_providers_dir / "claude" / "skills" + skills_dir.mkdir(parents=True) + outside = tmp_path / "outside-skill" + outside.mkdir() + (skills_dir / "python").symlink_to(outside, target_is_directory=True) + with pytest.raises(ValueError, match="symlinked skill directory"): + load_provider_skills("claude") + + def test_rejects_symlinked_skill_file(self, temp_providers_dir, tmp_path): + directory = temp_providers_dir / "claude" / "skills" / "python" + directory.mkdir(parents=True) + outside = tmp_path / "outside-skill.md" + outside.write_text("---\nname: python\ndescription: Test\n---\ncontent") + (directory / "SKILL.md").symlink_to(outside) + with pytest.raises(ValueError, match="symlinked SKILL.md"): + load_provider_skills("claude") + + +class TestSkillSetSelection: + def test_get_all_when_none(self): + skills = SkillSet("test", [ + Skill("a", "a-dir", "desc A", "content A"), + Skill("b", "b-dir", "desc B", "content B"), + ]) + result = skills.get(None) + assert len(result) == 2 + + def test_get_specific(self): + skills = SkillSet("test", [ + Skill("a", "a-dir", "desc A", "content A"), + Skill("b", "b-dir", "desc B", "content B"), + ]) + result = skills.get(["a"]) + assert len(result) == 1 + assert result[0].name == "a" + + def test_get_empty_list(self): + skills = SkillSet("test", [Skill("a", "a-dir", "desc", "c")]) + assert skills.get([]) == [] + + def test_warn_on_missing(self, capsys): + skills = SkillSet("test", [Skill("a", "a-dir", "desc", "c")]) + result = skills.get(["missing"], on_missing="warn") + assert result == [] + captured = capsys.readouterr() + assert "not found" in captured.err + + def test_error_on_missing(self): + skills = SkillSet("test", [Skill("a", "a-dir", "desc", "c")]) + with pytest.raises(KeyError, match="not found"): + skills.get(["missing"], on_missing="error") + + def test_ignore_on_missing(self): + skills = SkillSet("test", [Skill("a", "a-dir", "desc", "c")]) + result = skills.get(["missing"], on_missing="ignore") + assert result == [] + + def test_invalid_missing_policy(self): + skills = SkillSet("test") + with pytest.raises(ValueError, match="missing-skill policy"): + skills.get([], on_missing="silent") + + +class TestInjectSkills: + @pytest.fixture + def temp_providers_dir(self, tmp_path, monkeypatch): + monkeypatch.setattr("cmdforge.skills.PROVIDERS_DIR", tmp_path) + return tmp_path + + def test_no_skills_no_injection(self, temp_providers_dir): + d = temp_providers_dir / "empty" / "skills" + d.mkdir(parents=True) + result = inject_skills("Hello", "empty") + assert result == "Hello" + + def test_injects_skills_before_prompt(self, temp_providers_dir): + write_skill(temp_providers_dir, "claude", "python", "Use type hints.") + + result = inject_skills("Do X", "claude") + assert result.startswith("## Skill: python") + assert "Do X" in result + assert result.index("## Skill") < result.index("Do X") + + def test_specific_skill_names(self, temp_providers_dir): + for name in ("python", "git", "testing"): + write_skill(temp_providers_dir, "claude", name, f"{name} content") + + result = inject_skills("Prompt", "claude", ["python"]) + assert "python content" in result + assert "git content" not in result + assert "testing content" not in result + + def test_invalid_provider_name(self, temp_providers_dir): + with pytest.raises(ValueError, match="Invalid provider"): + inject_skills("Prompt", "../../etc") + + +class TestPromptStepSkills: + def test_to_dict_excludes_none(self): + step = PromptStep( + prompt="Test", + provider="claude", + output_var="out", + ) + d = step.to_dict() + assert "skills" not in d + + def test_to_dict_includes_skills(self): + step = PromptStep( + prompt="Test", + provider="claude", + output_var="out", + skills=["python", "git"], + ) + d = step.to_dict() + assert d["skills"] == ["python", "git"] + + def test_roundtrip_with_skills(self): + step = PromptStep( + prompt="Test", + provider="claude", + output_var="out", + skills=["python"], + ) + restored = PromptStep.from_dict(step.to_dict()) + assert restored.skills == ["python"] + + def test_roundtrip_without_skills(self): + step = PromptStep( + prompt="Test", + provider="claude", + output_var="out", + ) + restored = PromptStep.from_dict(step.to_dict()) + assert restored.skills is None + + def test_roundtrip_with_explicit_empty_skills(self): + step = PromptStep("Test", "claude", "out", skills=[]) + serialized = step.to_dict() + assert serialized["skills"] == [] + assert PromptStep.from_dict(serialized).skills == [] + + @pytest.mark.parametrize("skills", ["python", [1], ["*", "python"], ["python", "python"]]) + def test_rejects_invalid_skill_lists(self, skills): + with pytest.raises(ValueError): + PromptStep("Test", "claude", "out", skills=skills) + + def test_rejects_invalid_skill_identifier(self): + with pytest.raises(ValueError, match="Invalid skill name"): + PromptStep("Test", "claude", "out", skills=["Python_Utils"]) + + +class TestRunnerSkillIntegration: + @pytest.fixture + def configured_skill(self, tmp_path, monkeypatch): + monkeypatch.setattr("cmdforge.skills.PROVIDERS_DIR", tmp_path) + write_skill(tmp_path, "override", "python", "SKILL INSTRUCTIONS") + return tmp_path + + def test_plain_text_uses_effective_provider_and_context_order( + self, configured_skill, monkeypatch + ): + captured = {} + + def fake_call(provider, prompt, max_tokens=None): + captured.update(provider=provider, prompt=prompt) + return ProviderResult(text="done", success=True) + + monkeypatch.setattr("cmdforge.runner.call_provider", fake_call) + monkeypatch.setattr( + "cmdforge.runner.load_profile", + lambda name: SimpleNamespace(system_prompt="PROFILE INSTRUCTIONS"), + ) + step = PromptStep( + prompt="USER PROMPT", + provider="original", + output_var="out", + profile="reviewer", + plain_text=True, + skills=["python"], + ) + + output, success = execute_prompt_step( + step, {}, provider_override="override" + ) + + assert success is True + assert output == "done" + assert captured["provider"] == "override" + prompt = captured["prompt"] + assert prompt.index("PROFILE INSTRUCTIONS") < prompt.index("## Skill: python") + assert prompt.index("## Skill: python") < prompt.index("USER PROMPT") + + def test_structured_path_preserves_context_order( + self, configured_skill, monkeypatch + ): + captured = {} + + def fake_call(provider, prompt, max_tokens=None): + captured.update(provider=provider, prompt=prompt) + return ProviderResult(text='{"output": "done"}', success=True) + + monkeypatch.setattr("cmdforge.runner.call_provider", fake_call) + monkeypatch.setattr( + "cmdforge.runner.load_profile", + lambda name: SimpleNamespace(system_prompt="PROFILE INSTRUCTIONS"), + ) + step = PromptStep( + prompt="USER PROMPT", + provider="override", + output_var="out", + profile="reviewer", + skills=["python"], + ) + + output, success = execute_prompt_step(step, {}) + + assert success is True + assert output == {"output": "done"} + prompt = captured["prompt"] + assert prompt.index("PROFILE INSTRUCTIONS") < prompt.index("## Skill: python") + assert prompt.index("## Skill: python") < prompt.index("USER PROMPT")