Add M7.5 provider-attached skills system with Agent Skills spec compliance

This commit is contained in:
rob 2026-07-20 02:01:35 -03:00
parent fe53721725
commit 8d38bdd83a
4 changed files with 692 additions and 5 deletions

View File

@ -533,15 +533,22 @@ def execute_prompt_step(
prompt = substitute_variables(prompt_template, variables, warn_non_scalar=verbose) 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: if step.profile:
profile = load_profile(step.profile) profile = load_profile(step.profile)
if profile and profile.system_prompt: if profile and profile.system_prompt:
# Prepend system prompt to user prompt
prompt = f"{profile.system_prompt}\n\n---\n\n{prompt}" prompt = f"{profile.system_prompt}\n\n---\n\n{prompt}"
# Determine provider # Structured or plain-text
provider = provider_override or substitute_variables(step.provider, variables, warn_non_scalar=verbose)
# Plain text mode - bypass structured output # Plain text mode - bypass structured output
if step.plain_text: if step.plain_text:

221
src/cmdforge/skills.py Normal file
View File

@ -0,0 +1,221 @@
"""Provider-attached skills in the Agent Skills open standard.
Each provider can have skills at::
~/.cmdforge/providers/<provider>/skills/<skill-name>/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}"

View File

@ -130,6 +130,24 @@ class PromptStep:
max_retries: int = 1 # Retry count on validation failure max_retries: int = 1 # Retry count on validation failure
plain_text: bool = False # Bypass structured output enforcement plain_text: bool = False # Bypass structured output enforcement
max_tokens: Optional[int] = None # Max output tokens (provider-dependent) 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: def to_dict(self) -> dict:
d = { d = {
@ -154,6 +172,8 @@ class PromptStep:
d["plain_text"] = self.plain_text d["plain_text"] = self.plain_text
if self.max_tokens: if self.max_tokens:
d["max_tokens"] = self.max_tokens d["max_tokens"] = self.max_tokens
if self.skills is not None:
d["skills"] = self.skills
return d return d
@classmethod @classmethod
@ -178,7 +198,8 @@ class PromptStep:
output_schema=data.get("output_schema"), output_schema=data.get("output_schema"),
max_retries=data.get("max_retries", 1), max_retries=data.get("max_retries", 1),
plain_text=data.get("plain_text", False), plain_text=data.get("plain_text", False),
max_tokens=max_tokens max_tokens=max_tokens,
skills=data.get("skills"),
) )

438
tests/test_skills.py Normal file
View File

@ -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")