feat(skill): add skill domain model and resolver
CI / lint (push) Waiting to run
CI / typecheck (push) Waiting to run
CI / security (push) Waiting to run
CI / quality (push) Waiting to run
CI / unit_tests (push) Waiting to run
CI / integration_tests (push) Waiting to run
CI / coverage (push) Blocked by required conditions
CI / build (push) Waiting to run
CI / docker (push) Blocked by required conditions

This commit is contained in:
Jeff
2026-02-13 22:42:28 +00:00
parent eb1638a84e
commit 20321f21ee
7 changed files with 2153 additions and 0 deletions
+177
View File
@@ -0,0 +1,177 @@
"""ASV benchmarks for Skill domain model and resolution.
Measures the performance of:
- Skill model construction (Pydantic validation)
- Skill.as_cli_dict() serialization
- Skill.from_config() factory
- SkillResolver.resolve_tools() flattening
- SkillResolver.compute_capability_summary() aggregation
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.domain.models.core.skill import (
Skill,
SkillAgentSource,
SkillInclude,
SkillInlineTool,
SkillMcpSource,
SkillResolver,
)
from cleveragents.domain.models.core.tool import (
ToolCapability,
ToolSource,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.skill import (
Skill,
SkillAgentSource,
SkillInclude,
SkillInlineTool,
SkillMcpSource,
SkillResolver,
)
from cleveragents.domain.models.core.tool import (
ToolCapability,
ToolSource,
)
def _make_skill() -> Skill:
"""Create a fully-populated Skill for benchmarking."""
return Skill(
name="bench/full-skill",
description="Benchmark skill",
tool_refs=["bench/tool-a", "bench/tool-b", "bench/tool-c"],
includes=[SkillInclude(name="bench/base-skill")],
anonymous_tools=[
SkillInlineTool(
description="Inline bench tool",
source=ToolSource.CUSTOM,
code="return True",
capability=ToolCapability(writes=True, side_effects=["fs"]),
),
],
mcp_servers=[
SkillMcpSource(
server="npx @mcp/bench-server",
tools=["read", "write"],
env={"KEY": "val"},
),
],
agent_skills=[SkillAgentSource(path="./skills/bench")],
overrides={"bench/tool-a": {"timeout": 600}},
)
def _make_registry() -> dict[str, Skill]:
"""Create a skill registry for resolution benchmarks."""
base = Skill(
name="bench/base-skill",
description="Base skill",
tool_refs=["bench/base-tool-1", "bench/base-tool-2"],
)
return {"bench/base-skill": base}
class SkillConstructionSuite:
"""Benchmark Skill model construction (Pydantic validation)."""
def time_skill_construction(self) -> None:
"""Benchmark creating a fully-populated Skill object."""
_make_skill()
def time_skill_minimal_construction(self) -> None:
"""Benchmark creating a minimal Skill object."""
Skill(
name="bench/minimal",
description="Minimal skill",
)
def time_inline_tool_construction(self) -> None:
"""Benchmark SkillInlineTool construction."""
SkillInlineTool(
description="Bench inline",
source=ToolSource.CUSTOM,
code="return True",
)
def time_mcp_source_construction(self) -> None:
"""Benchmark SkillMcpSource construction."""
SkillMcpSource(
server="npx @mcp/bench",
tools=["read", "write"],
)
class SkillSerializationSuite:
"""Benchmark Skill serialization methods."""
def setup(self) -> None:
"""Create objects for serialization benchmarks."""
self.skill = _make_skill()
def time_skill_as_cli_dict(self) -> None:
"""Benchmark Skill.as_cli_dict() ordered dict generation."""
self.skill.as_cli_dict()
def time_skill_model_dump(self) -> None:
"""Benchmark Pydantic model_dump() serialization."""
self.skill.model_dump()
def time_skill_model_dump_json(self) -> None:
"""Benchmark Pydantic model_dump_json() JSON serialization."""
self.skill.model_dump_json()
class SkillFromConfigSuite:
"""Benchmark Skill.from_config() factory."""
def setup(self) -> None:
"""Prepare config dict for benchmarks."""
self.config = {
"name": "bench/from-config",
"description": "Config benchmark",
"tools": ["bench/tool-a", "bench/tool-b"],
"includes": [{"name": "bench/base-skill"}],
"anonymous_tools": [
{
"description": "Inline",
"source": "custom",
"code": "return True",
},
],
"mcp_servers": [
{"server": "npx @mcp/bench", "tools": ["read"]},
],
"agent_skills": [{"path": "./skills/bench"}],
}
def time_skill_from_config(self) -> None:
"""Benchmark Skill.from_config()."""
Skill.from_config(self.config)
class SkillResolutionSuite:
"""Benchmark SkillResolver.resolve_tools()."""
def setup(self) -> None:
"""Create skill and registry for benchmarks."""
self.skill = _make_skill()
self.registry = _make_registry()
self.registry[self.skill.name] = self.skill
self.resolver = SkillResolver()
def time_resolve_tools(self) -> None:
"""Benchmark resolve_tools() with one include level."""
self.resolver.resolve_tools(self.skill, self.registry)
def time_compute_capability_summary(self) -> None:
"""Benchmark compute_capability_summary()."""
resolved = self.resolver.resolve_tools(self.skill, self.registry)
self.resolver.compute_capability_summary(self.skill, resolved)
+106
View File
@@ -0,0 +1,106 @@
# Skill Domain Model
A `Skill` is a namespaced, reusable collection of tools in CleverAgents.
Skills are registered via YAML config and provide actors with a flattened
set of tools from multiple sources.
## Skill Name
Skills use a `namespace/short_name` naming pattern:
```
local/code-tools
devops/deploy-tools
```
The name is the unique identifier, validated against the pattern
`^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$`.
## Sub-models
### SkillToolRef
Reference to a named tool in the Tool Registry.
| Field | Type | Description |
|--------|-------|---------------------------------|
| `name` | `str` | Namespaced tool name reference |
### SkillInclude
Reference to another skill to include (recursive composition).
| Field | Type | Default | Description |
|-------------|--------------------|---------|--------------------------------|
| `name` | `str` | (req.) | Namespaced skill name |
| `overrides` | `dict \| None` | `None` | Per-tool metadata overrides |
### SkillInlineTool
Anonymous inline tool definition embedded in a skill.
| Field | Type | Default | Description |
|-----------------|-------------------------|---------|--------------------------------|
| `description` | `str` | (req.) | Description of the inline tool |
| `source` | `ToolSource` | (req.) | Source type |
| `code` | `str \| None` | `None` | Inline code |
| `input_schema` | `dict \| None` | `None` | JSON Schema for inputs |
| `output_schema` | `dict \| None` | `None` | JSON Schema for outputs |
| `capability` | `ToolCapability \| None`| `None` | Capability metadata |
| `resource_slots`| `list[ResourceSlot]` | `[]` | Resource slot declarations |
| `timeout` | `int` | `300` | Execution timeout (>= 1) |
### SkillMcpSource
MCP server source for a skill.
| Field | Type | Default | Description |
|----------|---------------------|---------|---------------------------------|
| `server` | `str` | (req.) | MCP server command or URI |
| `tools` | `list[str] \| None` | `None` | Specific tools (None = all) |
| `env` | `dict \| None` | `None` | Environment variables |
### SkillAgentSource
Agent Skills Standard folder source.
| Field | Type | Description |
|--------|-------|----------------------------------------|
| `path` | `str` | Path to Agent Skills Standard folder |
### SkillCapabilitySummary
Aggregated capability summary computed after resolution.
| Field | Type | Default | Description |
|------------------------|--------|---------|------------------------------------|
| `total_tools` | `int` | `0` | Total number of tools |
| `read_only_tools` | `int` | `0` | Number of read-only tools |
| `write_tools` | `int` | `0` | Number of write-capable tools |
| `checkpointable_tools` | `int` | `0` | Number of checkpointable tools |
| `has_side_effects` | `bool` | `False` | Whether any tool has side effects |
| `mcp_sources` | `int` | `0` | Number of MCP server sources |
| `agent_skill_sources` | `int` | `0` | Number of Agent Skill sources |
## Skill Fields
| Field | Type | Default | Description |
|-------------------|--------------------------------|---------|---------------------------------|
| `name` | `str` | (req.) | `namespace/short_name` format |
| `description` | `str` | (req.) | Short description |
| `tool_refs` | `list[str]` | `[]` | Named tool references |
| `includes` | `list[SkillInclude]` | `[]` | Included skills |
| `anonymous_tools` | `list[SkillInlineTool]` | `[]` | Inline tool definitions |
| `mcp_servers` | `list[SkillMcpSource]` | `[]` | MCP server sources |
| `agent_skills` | `list[SkillAgentSource]` | `[]` | Agent Skill folder sources |
| `overrides` | `dict[str, dict[str, Any]]` | `{}` | Per-tool metadata overrides |
## Class Methods
- `Skill.from_config(config: dict)` -- Create from a YAML config dict
- `skill.as_cli_dict()` -- Stable ordered dict for CLI rendering
## Properties
- `skill.namespace` -- Extracted namespace from name
- `skill.short_name` -- Extracted short name from name
+84
View File
@@ -0,0 +1,84 @@
# Skill Resolution
The `SkillResolver` flattens a skill's tool set by recursively resolving
includes, collecting tool references, inline tools, and external sources
into a deterministic ordered list.
## Resolution Algorithm
1. **Cycle detection** -- Before descending into includes, check if the
current skill has already been visited in this resolution path. If so,
raise a `ValueError` with the full cycle path trace.
2. **Depth-first include resolution** -- Process each `SkillInclude` by
recursively resolving the included skill first. This ensures that
tools from deeper includes appear earlier in the final order.
3. **Tool ref collection** -- Add tool_refs from the current skill. If a
tool name already exists (from an include), it is replaced (last-wins
semantics for overrides).
4. **Inline tool collection** -- Anonymous tools are keyed as
`{skill_name}/_anon_{index}` and added in definition order.
5. **MCP source collection** -- For MCP servers with explicit tool lists,
each tool is added as `mcp:{server}/{tool_name}`.
6. **Agent skill collection** -- Each agent skill path is added as
`agent_skill:{path}`.
## De-duplication
When the same tool name appears multiple times (e.g., from overlapping
includes), the **last occurrence wins**. The tool's position in the
ordered list is preserved from its first appearance, but the entry
metadata is updated.
## Override Application
Overrides defined in the skill's `overrides` dict are applied to matching
tool refs during resolution. The overrides dict is keyed by tool name and
contains arbitrary metadata that is attached to the `ResolvedToolEntry`.
## Cycle Detection
The resolver tracks the current resolution path (stack of skill names).
If a skill is encountered that is already on the path, a `ValueError` is
raised with a human-readable path trace:
```
Cycle detected in skill includes: local/a -> local/b -> local/a
```
## CapabilitySummary
After resolution, `SkillResolver.compute_capability_summary()` aggregates
capability metadata from inline tools to produce a `SkillCapabilitySummary`:
- Counts read-only, write, and checkpointable tools
- Detects side effects across all inline tools
- Counts MCP and agent skill sources
## API
```python
resolver = SkillResolver()
# Resolve tools
entries = resolver.resolve_tools(skill, skill_lookup_dict)
# Compute summary
summary = resolver.compute_capability_summary(skill, entries)
```
## ResolvedToolEntry
Each entry in the resolved list contains:
| Field | Type | Description |
|----------------|--------------------------|--------------------------------------|
| `name` | `str` | Tool name or generated key |
| `source_skill` | `str` | Skill that contributed this tool |
| `is_inline` | `bool` | Whether this is an inline tool |
| `inline_tool` | `SkillInlineTool \| None`| Inline definition (if applicable) |
| `overrides` | `dict[str, Any]` | Applied overrides |
+277
View File
@@ -0,0 +1,277 @@
Feature: Skill Domain Model and Resolution
As a developer
I want skill domain models and a resolver for the skill registry
So that skills can be registered, validated, resolved, and managed via CLI
# ---- Skill Creation ----
Scenario: Create a skill with valid fields
When I create a skill_model with name "local/code-tools" and description "Code editing tools"
Then the skill_model should be created
And the skill_model name should be "local/code-tools"
And the skill_model description should be "Code editing tools"
And the skill_model namespace should be "local"
And the skill_model short_name should be "code-tools"
Scenario: Create a skill with tool refs
When I create a skill_model with tool refs "local/edit-file,local/read-file"
Then the skill_model should be created
And the skill_model should have 2 tool refs
Scenario: Create a skill with all fields populated
When I create a skill_model with all fields
Then the skill_model should be created
And the skill_model should have 2 tool refs
And the skill_model should have 1 include
And the skill_model should have 1 anonymous tool
And the skill_model should have 1 mcp server
And the skill_model should have 1 agent skill
# ---- Skill Name Validation ----
Scenario: Skill name must match namespace/name pattern
When I try to create a skill_model with invalid name "no-namespace"
Then a skill_model validation error should be raised
And the skill_model error should mention "namespace/name"
Scenario: Skill name rejects spaces
When I try to create a skill_model with invalid name "local/bad name"
Then a skill_model validation error should be raised
Scenario: Skill name rejects double slashes
When I try to create a skill_model with invalid name "local//bad"
Then a skill_model validation error should be raised
Scenario: Skill name allows hyphens and underscores
When I create a skill_model with name "my-ns/my_skill" and description "Test"
Then the skill_model should be created
And the skill_model namespace should be "my-ns"
And the skill_model short_name should be "my_skill"
Scenario: Skill description must not be empty
When I try to create a skill_model with empty description
Then a skill_model validation error should be raised
# ---- Include Resolution (flat) ----
Scenario: Resolve skill with flat includes
Given a skill_resolver registry with "local/basic-tools" containing refs "local/edit-file,local/read-file"
And a skill_resolver root skill "local/code-tools" including "local/basic-tools" with own refs "local/format"
When I resolve the skill_resolver root skill
Then the skill_resolver result should have 3 tools
And the skill_resolver result should contain tool "local/edit-file"
And the skill_resolver result should contain tool "local/read-file"
And the skill_resolver result should contain tool "local/format"
# ---- Include Resolution (nested) ----
Scenario: Resolve skill with nested includes
Given a skill_resolver registry with "local/base" containing refs "local/core-tool"
And a skill_resolver registry with "local/mid" including "local/base" and containing refs "local/mid-tool"
And a skill_resolver root skill "local/top" including "local/mid" with own refs "local/top-tool"
When I resolve the skill_resolver root skill
Then the skill_resolver result should have 3 tools
And the skill_resolver result should contain tool "local/core-tool"
And the skill_resolver result should contain tool "local/mid-tool"
And the skill_resolver result should contain tool "local/top-tool"
# ---- Cycle Detection ----
Scenario: Cycle detection with error message
Given a skill_resolver registry with "local/a" including "local/b"
And a skill_resolver registry with "local/b" including "local/a"
When I try to resolve skill_resolver "local/a"
Then a skill_resolver cycle error should be raised
And the skill_resolver error should mention "local/a"
And the skill_resolver error should mention "local/b"
Scenario: Self-referencing cycle detection
Given a skill_resolver registry with "local/self" including "local/self"
When I try to resolve skill_resolver "local/self"
Then a skill_resolver cycle error should be raised
# ---- De-duplication ----
Scenario: De-duplication behavior last wins for overrides
Given a skill_resolver registry with "local/a" containing refs "local/shared-tool"
And a skill_resolver registry with "local/b" containing refs "local/shared-tool"
And a skill_resolver root skill "local/top" including "local/a,local/b" with no own refs
When I resolve the skill_resolver root skill
Then the skill_resolver result should have 1 tools
And the skill_resolver tool "local/shared-tool" should come from "local/b"
# ---- Override Application ----
Scenario: Override application on tool refs
When I create a skill_model with overrides for "local/edit-file"
And I resolve the overridden skill_model
Then the skill_resolver tool "local/edit-file" should have overrides
# ---- Empty Skill ----
Scenario: Empty skill with no tools or includes
When I create a skill_model with name "local/empty" and description "Empty skill"
And I resolve the empty skill_model
Then the skill_resolver result should have 0 tools
# ---- from_config Loading ----
Scenario: Load skill from config dict
When I load a skill_model from config with name "local/my-skill"
Then the skill_model should be created
And the skill_model name should be "local/my-skill"
Scenario: Load skill from config missing name raises error
When I try to load a skill_model from config missing "name"
Then a skill_model config error should be raised with "name"
Scenario: Load skill from config missing description raises error
When I try to load a skill_model from config missing "description"
Then a skill_model config error should be raised with "description"
Scenario: Load skill from config with full YAML structure
When I load a skill_model from full config
Then the skill_model should be created
And the skill_model should have 2 tool refs
And the skill_model should have 1 include
And the skill_model should have 1 anonymous tool
And the skill_model should have 1 mcp server
And the skill_model should have 1 agent skill
Scenario: Load skill from config with string includes
When I load a skill_model from config with string includes
Then the skill_model should be created
And the skill_model should have 1 include
Scenario: Load skill from config with string agent skills
When I load a skill_model from config with string agent skills
Then the skill_model should be created
And the skill_model should have 1 agent skill
# ---- as_cli_dict Output ----
Scenario: Skill as_cli_dict has required keys
When I create a skill_model and call as_cli_dict
Then the skill_model cli dict should have key "name"
And the skill_model cli dict should have key "description"
And the skill_model cli dict should have key "namespace"
And the skill_model cli dict should have key "short_name"
Scenario: Skill as_cli_dict includes optional sections when set
When I create a skill_model with all fields and call as_cli_dict
Then the skill_model cli dict should have key "tool_refs"
And the skill_model cli dict should have key "includes"
And the skill_model cli dict should have key "anonymous_tools"
And the skill_model cli dict should have key "mcp_servers"
And the skill_model cli dict should have key "agent_skills"
And the skill_model cli dict should have key "overrides"
Scenario: Skill as_cli_dict omits empty sections
When I create a skill_model with name "local/minimal" and description "Minimal"
And I get the skill_model cli dict
Then the skill_model cli dict should not have key "tool_refs"
And the skill_model cli dict should not have key "includes"
And the skill_model cli dict should not have key "anonymous_tools"
And the skill_model cli dict should not have key "mcp_servers"
And the skill_model cli dict should not have key "agent_skills"
And the skill_model cli dict should not have key "overrides"
# ---- Inline Tool Validation ----
Scenario: Inline tool with valid custom source
When I create a skill_model inline tool with source "custom" and code "print('hi')"
Then the skill_model inline tool should be created
Scenario: Inline tool has default timeout of 300
When I create a skill_model inline tool with source "custom" and code "print('hi')"
Then the skill_model inline tool timeout should be 300
Scenario: Inline tool timeout must be >= 1
When I try to create a skill_model inline tool with timeout 0
Then a skill_model validation error should be raised
Scenario: Inline tool description must not be empty
When I try to create a skill_model inline tool with empty description
Then a skill_model validation error should be raised
# ---- MCP Source Validation ----
Scenario: MCP source with specific tools
When I create a skill_model mcp source with server "npx @mcp/server" and tools "read,write"
Then the skill_model mcp source should be created
And the skill_model mcp source should have 2 tools
Scenario: MCP source with all tools (none specified)
When I create a skill_model mcp source with server "npx @mcp/server" and no tools
Then the skill_model mcp source should be created
And the skill_model mcp source tools should be none
Scenario: MCP source with env vars
When I create a skill_model mcp source with env vars
Then the skill_model mcp source should be created
Scenario: MCP source server must not be empty
When I try to create a skill_model mcp source with empty server
Then a skill_model validation error should be raised
# ---- Agent Skill Source Validation ----
Scenario: Agent skill source with valid path
When I create a skill_model agent source with path "./skills/review"
Then the skill_model agent source should be created
Scenario: Agent skill source path must not be empty
When I try to create a skill_model agent source with empty path
Then a skill_model validation error should be raised
# ---- CapabilitySummary Computation ----
Scenario: CapabilitySummary computation for skill with inline tools
When I compute skill_model capability summary for a skill with inline tools
Then the skill_model summary total tools should be 4
And the skill_model summary read_only_tools should be 1
And the skill_model summary write_tools should be 1
And the skill_model summary has_side_effects should be true
And the skill_model summary mcp_sources should be 1
And the skill_model summary agent_skill_sources should be 1
Scenario: CapabilitySummary for empty skill
When I compute skill_model capability summary for an empty skill
Then the skill_model summary total tools should be 0
# ---- SkillToolRef ----
Scenario: SkillToolRef creation
When I create a skill_model tool ref with name "local/my-tool"
Then the skill_model tool ref should be created
# ---- SkillInclude ----
Scenario: SkillInclude creation with overrides
When I create a skill_model include with name "local/base" and overrides
Then the skill_model include should be created
And the skill_model include should have overrides
Scenario: SkillInclude creation without overrides
When I create a skill_model include with name "local/base" without overrides
Then the skill_model include should be created
# ---- Missing included skill ----
Scenario: Resolve with missing included skill raises error
Given a skill_resolver root skill "local/broken" including "local/nonexistent" with no own refs
When I try to resolve skill_resolver "local/broken"
Then a skill_resolver missing skill error should be raised
And the skill_resolver error should mention "local/nonexistent"
# ---- MCP and agent skill resolution ----
Scenario: Resolve skill with MCP and agent skill sources
When I create and resolve a skill_model with mcp and agent sources
Then the skill_resolver result should have 3 tools
# ---- ResolvedToolEntry ----
Scenario: ResolvedToolEntry creation
When I create a skill_resolver resolved tool entry
Then the skill_resolver entry should be created
+922
View File
@@ -0,0 +1,922 @@
"""Step definitions for Skill domain model and resolution tests."""
from typing import Any
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.skill import (
ResolvedToolEntry,
Skill,
SkillAgentSource,
SkillInclude,
SkillInlineTool,
SkillMcpSource,
SkillResolver,
SkillToolRef,
)
from cleveragents.domain.models.core.tool import (
ToolCapability,
ToolSource,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_skill(**overrides: Any) -> Skill:
"""Create a Skill with sensible defaults, allowing overrides."""
defaults: dict[str, Any] = {
"name": "local/test-skill",
"description": "A test skill",
}
defaults.update(overrides)
return Skill(**defaults)
def _skill_config(**overrides: Any) -> dict[str, Any]:
"""Return a minimal valid config dict for Skill.from_config."""
cfg: dict[str, Any] = {
"name": "local/my-skill",
"description": "A skill from config",
}
cfg.update(overrides)
return cfg
# ---------------------------------------------------------------------------
# Skill Creation Steps
# ---------------------------------------------------------------------------
@when('I create a skill_model with name "{name}" and description "{desc}"')
def skill_model_create_basic(context: Context, name: str, desc: str) -> None:
"""Create a skill with name and description."""
context.skill_model = Skill(name=name, description=desc)
context.skill_model_error = None
@when('I create a skill_model with tool refs "{refs}"')
def skill_model_create_with_refs(context: Context, refs: str) -> None:
"""Create a skill with tool refs."""
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
context.skill_model = _make_skill(tool_refs=ref_list)
context.skill_model_error = None
@when("I create a skill_model with all fields")
def skill_model_create_all_fields(context: Context) -> None:
"""Create a skill with all fields populated."""
context.skill_model = Skill(
name="local/full-skill",
description="Full skill",
tool_refs=["local/edit-file", "local/read-file"],
includes=[SkillInclude(name="local/basic-tools")],
anonymous_tools=[
SkillInlineTool(
description="Inline helper",
source=ToolSource.CUSTOM,
code="print('hello')",
),
],
mcp_servers=[
SkillMcpSource(
server="npx @mcp/server-fs",
tools=["read_file", "write_file"],
),
],
agent_skills=[SkillAgentSource(path="./skills/code-review")],
overrides={"local/edit-file": {"timeout": 600}},
)
context.skill_model_error = None
# ---------------------------------------------------------------------------
# Skill Assertions
# ---------------------------------------------------------------------------
@then("the skill_model should be created")
def skill_model_check_created(context: Context) -> None:
"""Verify the skill was created."""
assert context.skill_model is not None, "Skill should be created"
@then('the skill_model name should be "{expected}"')
def skill_model_check_name(context: Context, expected: str) -> None:
"""Check skill name."""
assert context.skill_model.name == expected, (
f"Expected name '{expected}', got '{context.skill_model.name}'"
)
@then('the skill_model description should be "{expected}"')
def skill_model_check_description(context: Context, expected: str) -> None:
"""Check skill description."""
assert context.skill_model.description == expected
@then('the skill_model namespace should be "{expected}"')
def skill_model_check_namespace(context: Context, expected: str) -> None:
"""Check skill namespace property."""
assert context.skill_model.namespace == expected
@then('the skill_model short_name should be "{expected}"')
def skill_model_check_short_name(context: Context, expected: str) -> None:
"""Check skill short_name property."""
assert context.skill_model.short_name == expected
@then("the skill_model should have {count:d} tool refs")
def skill_model_check_ref_count(context: Context, count: int) -> None:
"""Check skill tool_refs count."""
actual = len(context.skill_model.tool_refs)
assert actual == count, f"Expected {count} tool refs, got {actual}"
@then("the skill_model should have {count:d} include")
def skill_model_check_include_count(context: Context, count: int) -> None:
"""Check skill includes count."""
actual = len(context.skill_model.includes)
assert actual == count, f"Expected {count} includes, got {actual}"
@then("the skill_model should have {count:d} anonymous tool")
def skill_model_check_anon_count(context: Context, count: int) -> None:
"""Check skill anonymous_tools count."""
actual = len(context.skill_model.anonymous_tools)
assert actual == count, f"Expected {count} anonymous tools, got {actual}"
@then("the skill_model should have {count:d} mcp server")
def skill_model_check_mcp_count(context: Context, count: int) -> None:
"""Check skill mcp_servers count."""
actual = len(context.skill_model.mcp_servers)
assert actual == count, f"Expected {count} mcp servers, got {actual}"
@then("the skill_model should have {count:d} agent skill")
def skill_model_check_agent_count(context: Context, count: int) -> None:
"""Check skill agent_skills count."""
actual = len(context.skill_model.agent_skills)
assert actual == count, f"Expected {count} agent skills, got {actual}"
# ---------------------------------------------------------------------------
# Name Validation
# ---------------------------------------------------------------------------
@when('I try to create a skill_model with invalid name "{name}"')
def skill_model_try_invalid_name(context: Context, name: str) -> None:
"""Attempt to create a skill with an invalid name."""
context.skill_model_error = None
context.skill_model = None
try:
context.skill_model = _make_skill(name=name)
except ValidationError as e:
context.skill_model_error = e
@when("I try to create a skill_model with empty description")
def skill_model_try_empty_desc(context: Context) -> None:
"""Attempt to create a skill with empty description."""
context.skill_model_error = None
context.skill_model = None
try:
context.skill_model = Skill(name="local/bad", description="")
except ValidationError as e:
context.skill_model_error = e
@then("a skill_model validation error should be raised")
def skill_model_check_validation_error(context: Context) -> None:
"""Verify that a validation error was raised."""
assert context.skill_model_error is not None, (
"Expected a validation error to be raised"
)
@then('the skill_model error should mention "{text}"')
def skill_model_check_error_message(context: Context, text: str) -> None:
"""Check the error message contains expected text."""
error_str = str(context.skill_model_error)
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
# ---------------------------------------------------------------------------
# Include Resolution (flat)
# ---------------------------------------------------------------------------
@given('a skill_resolver registry with "{name}" containing refs "{refs}"')
def skill_resolver_registry_with_refs(context: Context, name: str, refs: str) -> None:
"""Set up a registry with a skill containing tool refs."""
if not hasattr(context, "skill_registry"):
context.skill_registry = {}
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
context.skill_registry[name] = Skill(
name=name,
description=f"Skill {name}",
tool_refs=ref_list,
)
@given(
'a skill_resolver root skill "{name}" including "{includes}" with own refs "{refs}"'
)
def skill_resolver_root_with_includes(
context: Context, name: str, includes: str, refs: str
) -> None:
"""Set up the root skill to resolve."""
if not hasattr(context, "skill_registry"):
context.skill_registry = {}
include_list = [
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
]
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
context.skill_root = Skill(
name=name,
description=f"Root skill {name}",
tool_refs=ref_list,
includes=include_list,
)
context.skill_registry[name] = context.skill_root
@given('a skill_resolver root skill "{name}" including "{includes}" with no own refs')
def skill_resolver_root_no_refs(context: Context, name: str, includes: str) -> None:
"""Set up the root skill with includes and no own refs."""
if not hasattr(context, "skill_registry"):
context.skill_registry = {}
include_list = [
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
]
context.skill_root = Skill(
name=name,
description=f"Root skill {name}",
includes=include_list,
)
context.skill_registry[name] = context.skill_root
@when("I resolve the skill_resolver root skill")
def skill_resolver_resolve_root(context: Context) -> None:
"""Resolve the root skill."""
resolver = SkillResolver()
context.skill_resolved = resolver.resolve_tools(
context.skill_root, context.skill_registry
)
context.skill_resolver_error = None
@then("the skill_resolver result should have {count:d} tools")
def skill_resolver_check_tool_count(context: Context, count: int) -> None:
"""Check resolved tool count."""
actual = len(context.skill_resolved)
assert actual == count, (
f"Expected {count} tools, got {actual}: "
f"{[e.name for e in context.skill_resolved]}"
)
@then('the skill_resolver result should contain tool "{name}"')
def skill_resolver_check_contains_tool(context: Context, name: str) -> None:
"""Check resolved tools contain a specific tool."""
names = [e.name for e in context.skill_resolved]
assert name in names, f"Expected tool '{name}' in {names}"
# ---------------------------------------------------------------------------
# Include Resolution (nested)
# ---------------------------------------------------------------------------
@given(
'a skill_resolver registry with "{name}" including "{includes}" and containing refs "{refs}"'
)
def skill_resolver_registry_with_includes_and_refs(
context: Context, name: str, includes: str, refs: str
) -> None:
"""Set up a registry skill with includes and refs."""
if not hasattr(context, "skill_registry"):
context.skill_registry = {}
include_list = [
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
]
ref_list = [r.strip() for r in refs.split(",") if r.strip()]
context.skill_registry[name] = Skill(
name=name,
description=f"Skill {name}",
tool_refs=ref_list,
includes=include_list,
)
# ---------------------------------------------------------------------------
# Cycle Detection
# ---------------------------------------------------------------------------
@given('a skill_resolver registry with "{name}" including "{includes}"')
def skill_resolver_registry_with_includes(
context: Context, name: str, includes: str
) -> None:
"""Set up a registry skill that includes another."""
if not hasattr(context, "skill_registry"):
context.skill_registry = {}
include_list = [
SkillInclude(name=i.strip()) for i in includes.split(",") if i.strip()
]
context.skill_registry[name] = Skill(
name=name,
description=f"Skill {name}",
includes=include_list,
)
@when('I try to resolve skill_resolver "{name}"')
def skill_resolver_try_resolve(context: Context, name: str) -> None:
"""Try to resolve a skill and capture any error."""
if not hasattr(context, "skill_registry"):
context.skill_registry = {}
# If root skill is set in registry, use it; otherwise find it
skill = context.skill_registry.get(name)
if skill is None:
# Must be set via root skill step
skill = getattr(context, "skill_root", None)
assert skill is not None, f"Skill '{name}' not found in registry or root"
resolver = SkillResolver()
context.skill_resolver_error = None
context.skill_resolved = None
try:
context.skill_resolved = resolver.resolve_tools(skill, context.skill_registry)
except ValueError as e:
context.skill_resolver_error = e
@then("a skill_resolver cycle error should be raised")
def skill_resolver_check_cycle_error(context: Context) -> None:
"""Verify that a cycle error was raised."""
assert context.skill_resolver_error is not None, (
"Expected a cycle error to be raised"
)
assert "ycle" in str(context.skill_resolver_error), (
f"Expected cycle error, got: {context.skill_resolver_error}"
)
@then('the skill_resolver error should mention "{text}"')
def skill_resolver_check_error_mention(context: Context, text: str) -> None:
"""Check the resolver error mentions text."""
error_str = str(context.skill_resolver_error)
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
# ---------------------------------------------------------------------------
# De-duplication
# ---------------------------------------------------------------------------
@then('the skill_resolver tool "{name}" should come from "{source}"')
def skill_resolver_check_tool_source(context: Context, name: str, source: str) -> None:
"""Check which skill contributed a tool."""
for entry in context.skill_resolved:
if entry.name == name:
assert entry.source_skill == source, (
f"Expected tool '{name}' from '{source}', got '{entry.source_skill}'"
)
return
raise AssertionError(
f"Tool '{name}' not found in resolved: "
f"{[e.name for e in context.skill_resolved]}"
)
# ---------------------------------------------------------------------------
# Override Application
# ---------------------------------------------------------------------------
@when('I create a skill_model with overrides for "{tool_name}"')
def skill_model_create_with_overrides(context: Context, tool_name: str) -> None:
"""Create a skill with per-tool overrides."""
context.skill_model = Skill(
name="local/override-skill",
description="Skill with overrides",
tool_refs=[tool_name],
overrides={tool_name: {"timeout": 600}},
)
context.skill_model_error = None
@when("I resolve the overridden skill_model")
def skill_model_resolve_overridden(context: Context) -> None:
"""Resolve a skill with overrides."""
resolver = SkillResolver()
context.skill_resolved = resolver.resolve_tools(context.skill_model, {})
@then('the skill_resolver tool "{name}" should have overrides')
def skill_resolver_check_tool_overrides(context: Context, name: str) -> None:
"""Check that a resolved tool has overrides."""
for entry in context.skill_resolved:
if entry.name == name:
assert entry.overrides, f"Expected overrides on tool '{name}', got empty"
return
raise AssertionError(f"Tool '{name}' not found in resolved")
# ---------------------------------------------------------------------------
# Empty Skill
# ---------------------------------------------------------------------------
@when("I resolve the empty skill_model")
def skill_model_resolve_empty(context: Context) -> None:
"""Resolve an empty skill."""
resolver = SkillResolver()
context.skill_resolved = resolver.resolve_tools(context.skill_model, {})
# ---------------------------------------------------------------------------
# from_config Loading
# ---------------------------------------------------------------------------
@when('I load a skill_model from config with name "{name}"')
def skill_model_load_from_config(context: Context, name: str) -> None:
"""Load a skill from a config dict."""
config = _skill_config(name=name)
context.skill_model = Skill.from_config(config)
context.skill_model_error = None
@when('I try to load a skill_model from config missing "{field}"')
def skill_model_try_load_missing(context: Context, field: str) -> None:
"""Try loading a skill config with missing field."""
config = _skill_config()
del config[field]
context.skill_model_config_error = None
try:
Skill.from_config(config)
except ValueError as e:
context.skill_model_config_error = e
@then('a skill_model config error should be raised with "{field}"')
def skill_model_check_config_error(context: Context, field: str) -> None:
"""Check that a config error mentioning field was raised."""
assert context.skill_model_config_error is not None, (
f"Expected ValueError for missing '{field}'"
)
assert field in str(context.skill_model_config_error), (
f"Expected error to mention '{field}', got: {context.skill_model_config_error}"
)
@when("I load a skill_model from full config")
def skill_model_load_full_config(context: Context) -> None:
"""Load a skill from a full config with all sections."""
config = {
"name": "local/full-config",
"description": "Full config skill",
"tools": ["local/edit-file", "local/read-file"],
"includes": [{"name": "local/basic-tools"}],
"anonymous_tools": [
{
"description": "Inline helper",
"source": "custom",
"code": "print('hi')",
},
],
"mcp_servers": [
{
"server": "npx @mcp/server-fs",
"tools": ["read_file"],
},
],
"agent_skills": [{"path": "./skills/review"}],
"overrides": {"local/edit-file": {"timeout": 600}},
}
context.skill_model = Skill.from_config(config)
context.skill_model_error = None
@when("I load a skill_model from config with string includes")
def skill_model_load_config_string_includes(context: Context) -> None:
"""Load a skill from config where includes are plain strings."""
config = {
"name": "local/string-inc",
"description": "String includes",
"includes": ["local/basic-tools"],
}
context.skill_model = Skill.from_config(config)
context.skill_model_error = None
@when("I load a skill_model from config with string agent skills")
def skill_model_load_config_string_agents(context: Context) -> None:
"""Load a skill from config where agent_skills are plain strings."""
config = {
"name": "local/string-agents",
"description": "String agents",
"agent_skills": ["./skills/review"],
}
context.skill_model = Skill.from_config(config)
context.skill_model_error = None
# ---------------------------------------------------------------------------
# as_cli_dict
# ---------------------------------------------------------------------------
@when("I create a skill_model and call as_cli_dict")
def skill_model_create_and_cli_dict(context: Context) -> None:
"""Create a skill and get CLI dict."""
context.skill_model = _make_skill()
context.skill_cli_dict = context.skill_model.as_cli_dict()
@when("I create a skill_model with all fields and call as_cli_dict")
def skill_model_create_all_cli_dict(context: Context) -> None:
"""Create a skill with all fields and get CLI dict."""
context.skill_model = Skill(
name="local/full-cli",
description="Full CLI skill",
tool_refs=["local/edit-file"],
includes=[SkillInclude(name="local/base", overrides={"x": 1})],
anonymous_tools=[
SkillInlineTool(
description="Inline",
source=ToolSource.CUSTOM,
code="x = 1",
),
],
mcp_servers=[
SkillMcpSource(server="npx @mcp/fs", tools=["read"], env={"K": "V"}),
],
agent_skills=[SkillAgentSource(path="./skills/r")],
overrides={"local/edit-file": {"timeout": 600}},
)
context.skill_cli_dict = context.skill_model.as_cli_dict()
@when("I get the skill_model cli dict")
def skill_model_get_cli_dict(context: Context) -> None:
"""Get CLI dict from the current skill model."""
context.skill_cli_dict = context.skill_model.as_cli_dict()
@then('the skill_model cli dict should have key "{key}"')
def skill_model_check_cli_dict_key(context: Context, key: str) -> None:
"""Check CLI dict has expected key."""
assert key in context.skill_cli_dict, (
f"Expected key '{key}' in cli dict, keys: {list(context.skill_cli_dict)}"
)
@then('the skill_model cli dict should not have key "{key}"')
def skill_model_check_cli_dict_no_key(context: Context, key: str) -> None:
"""Check CLI dict does not have a key."""
assert key not in context.skill_cli_dict, f"Unexpected key '{key}' in cli dict"
# ---------------------------------------------------------------------------
# Inline Tool Validation
# ---------------------------------------------------------------------------
@when('I create a skill_model inline tool with source "{source}" and code "{code}"')
def skill_model_create_inline_tool(context: Context, source: str, code: str) -> None:
"""Create a SkillInlineTool."""
context.skill_inline_tool = SkillInlineTool(
description="Inline test tool",
source=ToolSource(source),
code=code,
)
context.skill_model_error = None
@then("the skill_model inline tool should be created")
def skill_model_check_inline_created(context: Context) -> None:
"""Verify inline tool was created."""
assert context.skill_inline_tool is not None
@then("the skill_model inline tool timeout should be {expected:d}")
def skill_model_check_inline_timeout(context: Context, expected: int) -> None:
"""Check inline tool timeout."""
assert context.skill_inline_tool.timeout == expected
@when("I try to create a skill_model inline tool with timeout {timeout:d}")
def skill_model_try_inline_bad_timeout(context: Context, timeout: int) -> None:
"""Attempt to create an inline tool with invalid timeout."""
context.skill_model_error = None
try:
SkillInlineTool(
description="Bad timeout",
source=ToolSource.CUSTOM,
code="x = 1",
timeout=timeout,
)
except ValidationError as e:
context.skill_model_error = e
@when("I try to create a skill_model inline tool with empty description")
def skill_model_try_inline_empty_desc(context: Context) -> None:
"""Attempt to create an inline tool with empty description."""
context.skill_model_error = None
try:
SkillInlineTool(
description="",
source=ToolSource.CUSTOM,
code="x = 1",
)
except ValidationError as e:
context.skill_model_error = e
# ---------------------------------------------------------------------------
# MCP Source Validation
# ---------------------------------------------------------------------------
@when('I create a skill_model mcp source with server "{server}" and tools "{tools}"')
def skill_model_create_mcp_source(context: Context, server: str, tools: str) -> None:
"""Create a SkillMcpSource with tools."""
tool_list = [t.strip() for t in tools.split(",") if t.strip()]
context.skill_mcp_source = SkillMcpSource(server=server, tools=tool_list)
context.skill_model_error = None
@when('I create a skill_model mcp source with server "{server}" and no tools')
def skill_model_create_mcp_no_tools(context: Context, server: str) -> None:
"""Create a SkillMcpSource without specific tools (all tools)."""
context.skill_mcp_source = SkillMcpSource(server=server)
context.skill_model_error = None
@when("I create a skill_model mcp source with env vars")
def skill_model_create_mcp_env(context: Context) -> None:
"""Create a SkillMcpSource with env vars."""
context.skill_mcp_source = SkillMcpSource(
server="npx @mcp/server",
env={"API_KEY": "test123"},
)
context.skill_model_error = None
@when("I try to create a skill_model mcp source with empty server")
def skill_model_try_mcp_empty_server(context: Context) -> None:
"""Attempt to create MCP source with empty server."""
context.skill_model_error = None
try:
SkillMcpSource(server="")
except ValidationError as e:
context.skill_model_error = e
@then("the skill_model mcp source should be created")
def skill_model_check_mcp_created(context: Context) -> None:
"""Verify MCP source was created."""
assert context.skill_mcp_source is not None
@then("the skill_model mcp source should have {count:d} tools")
def skill_model_check_mcp_tool_count(context: Context, count: int) -> None:
"""Check MCP source tool count."""
assert context.skill_mcp_source.tools is not None
actual = len(context.skill_mcp_source.tools)
assert actual == count, f"Expected {count} tools, got {actual}"
@then("the skill_model mcp source tools should be none")
def skill_model_check_mcp_tools_none(context: Context) -> None:
"""Check MCP source tools is None."""
assert context.skill_mcp_source.tools is None
# ---------------------------------------------------------------------------
# Agent Skill Source Validation
# ---------------------------------------------------------------------------
@when('I create a skill_model agent source with path "{path}"')
def skill_model_create_agent_source(context: Context, path: str) -> None:
"""Create a SkillAgentSource."""
context.skill_agent_source = SkillAgentSource(path=path)
context.skill_model_error = None
@when("I try to create a skill_model agent source with empty path")
def skill_model_try_agent_empty_path(context: Context) -> None:
"""Attempt to create agent source with empty path."""
context.skill_model_error = None
try:
SkillAgentSource(path="")
except ValidationError as e:
context.skill_model_error = e
@then("the skill_model agent source should be created")
def skill_model_check_agent_created(context: Context) -> None:
"""Verify agent source was created."""
assert context.skill_agent_source is not None
# ---------------------------------------------------------------------------
# CapabilitySummary Computation
# ---------------------------------------------------------------------------
@when("I compute skill_model capability summary for a skill with inline tools")
def skill_model_compute_summary(context: Context) -> None:
"""Compute capability summary for a skill with various tools."""
skill = Skill(
name="local/summary-skill",
description="Summary test",
tool_refs=["local/some-ref"],
anonymous_tools=[
SkillInlineTool(
description="Read-only tool",
source=ToolSource.CUSTOM,
code="x = 1",
capability=ToolCapability(read_only=True),
),
SkillInlineTool(
description="Write tool",
source=ToolSource.CUSTOM,
code="x = 2",
capability=ToolCapability(
writes=True,
side_effects=["filesystem"],
),
),
],
mcp_servers=[SkillMcpSource(server="npx @mcp/fs")],
agent_skills=[SkillAgentSource(path="./skills/r")],
)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
context.skill_summary = resolver.compute_capability_summary(skill, resolved)
@when("I compute skill_model capability summary for an empty skill")
def skill_model_compute_empty_summary(context: Context) -> None:
"""Compute capability summary for an empty skill."""
skill = Skill(
name="local/empty-summary",
description="Empty summary test",
)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
context.skill_summary = resolver.compute_capability_summary(skill, resolved)
@then("the skill_model summary total tools should be {expected:d}")
def skill_model_check_summary_total(context: Context, expected: int) -> None:
"""Check summary total_tools."""
assert context.skill_summary.total_tools == expected
@then("the skill_model summary read_only_tools should be {expected:d}")
def skill_model_check_summary_readonly(context: Context, expected: int) -> None:
"""Check summary read_only_tools."""
assert context.skill_summary.read_only_tools == expected
@then("the skill_model summary write_tools should be {expected:d}")
def skill_model_check_summary_writes(context: Context, expected: int) -> None:
"""Check summary write_tools."""
assert context.skill_summary.write_tools == expected
@then("the skill_model summary has_side_effects should be true")
def skill_model_check_summary_side_effects(context: Context) -> None:
"""Check summary has_side_effects."""
assert context.skill_summary.has_side_effects is True
@then("the skill_model summary mcp_sources should be {expected:d}")
def skill_model_check_summary_mcp(context: Context, expected: int) -> None:
"""Check summary mcp_sources."""
assert context.skill_summary.mcp_sources == expected
@then("the skill_model summary agent_skill_sources should be {expected:d}")
def skill_model_check_summary_agents(context: Context, expected: int) -> None:
"""Check summary agent_skill_sources."""
assert context.skill_summary.agent_skill_sources == expected
# ---------------------------------------------------------------------------
# SkillToolRef
# ---------------------------------------------------------------------------
@when('I create a skill_model tool ref with name "{name}"')
def skill_model_create_tool_ref(context: Context, name: str) -> None:
"""Create a SkillToolRef."""
context.skill_tool_ref = SkillToolRef(name=name)
@then("the skill_model tool ref should be created")
def skill_model_check_tool_ref(context: Context) -> None:
"""Verify tool ref was created."""
assert context.skill_tool_ref is not None
# ---------------------------------------------------------------------------
# SkillInclude
# ---------------------------------------------------------------------------
@when('I create a skill_model include with name "{name}" and overrides')
def skill_model_create_include_with_overrides(context: Context, name: str) -> None:
"""Create a SkillInclude with overrides."""
context.skill_include = SkillInclude(name=name, overrides={"timeout": 600})
@when('I create a skill_model include with name "{name}" without overrides')
def skill_model_create_include_no_overrides(context: Context, name: str) -> None:
"""Create a SkillInclude without overrides."""
context.skill_include = SkillInclude(name=name)
@then("the skill_model include should be created")
def skill_model_check_include_created(context: Context) -> None:
"""Verify include was created."""
assert context.skill_include is not None
@then("the skill_model include should have overrides")
def skill_model_check_include_overrides(context: Context) -> None:
"""Check include has overrides."""
assert context.skill_include.overrides is not None
# ---------------------------------------------------------------------------
# Missing included skill
# ---------------------------------------------------------------------------
@then("a skill_resolver missing skill error should be raised")
def skill_resolver_check_missing_error(context: Context) -> None:
"""Verify a missing skill error was raised."""
assert context.skill_resolver_error is not None, "Expected a missing skill error"
assert "not found" in str(context.skill_resolver_error), (
f"Expected 'not found' in error, got: {context.skill_resolver_error}"
)
# ---------------------------------------------------------------------------
# MCP and agent skill resolution
# ---------------------------------------------------------------------------
@when("I create and resolve a skill_model with mcp and agent sources")
def skill_model_resolve_mcp_agent(context: Context) -> None:
"""Create and resolve a skill with MCP and agent sources."""
skill = Skill(
name="local/mcp-agent",
description="MCP and agent test",
tool_refs=["local/base-tool"],
mcp_servers=[
SkillMcpSource(
server="npx @mcp/fs",
tools=["read_file"],
),
],
agent_skills=[SkillAgentSource(path="./skills/review")],
)
resolver = SkillResolver()
context.skill_resolved = resolver.resolve_tools(skill, {})
# ---------------------------------------------------------------------------
# ResolvedToolEntry
# ---------------------------------------------------------------------------
@when("I create a skill_resolver resolved tool entry")
def skill_resolver_create_entry(context: Context) -> None:
"""Create a ResolvedToolEntry."""
context.skill_resolved_entry = ResolvedToolEntry(
name="local/test-tool",
source_skill="local/my-skill",
)
@then("the skill_resolver entry should be created")
def skill_resolver_check_entry(context: Context) -> None:
"""Verify entry was created."""
assert context.skill_resolved_entry is not None
assert context.skill_resolved_entry.name == "local/test-tool"
assert context.skill_resolved_entry.is_inline is False
@@ -82,6 +82,19 @@ from cleveragents.domain.models.core.session import (
SessionTokenUsage,
)
# Skill domain models
from cleveragents.domain.models.core.skill import (
ResolvedToolEntry,
Skill,
SkillAgentSource,
SkillCapabilitySummary,
SkillInclude,
SkillInlineTool,
SkillMcpSource,
SkillResolver,
SkillToolRef,
)
# Tool and Validation domain models
from cleveragents.domain.models.core.tool import (
BindingMode,
@@ -142,6 +155,7 @@ __all__ = [
"ProjectLink",
"ProjectSettings",
"ProjectStats",
"ResolvedToolEntry",
"Resource",
"ResourceAccessMode",
"ResourceCapabilities",
@@ -155,6 +169,14 @@ __all__ = [
"SessionService",
"SessionServiceError",
"SessionTokenUsage",
"Skill",
"SkillAgentSource",
"SkillCapabilitySummary",
"SkillInclude",
"SkillInlineTool",
"SkillMcpSource",
"SkillResolver",
"SkillToolRef",
"SummaryForUpdateContextParams",
"TemporalScope",
"Tool",
@@ -0,0 +1,565 @@
"""Skill domain model and resolver for CleverAgents v3.
A Skill is a namespaced, reusable collection of tools that is registered via
YAML config. Skills reference named tools (from the Tool Registry), define
anonymous inline tools, include other skills, or expose MCP/Agent Skills.
When an actor references a skill it gains access to the **flattened set of all
tools** -- resolved recursively from includes, tool_refs, inline tools, and
external sources.
Skills are NOT single tools -- they are containers/collections.
Namespaced as ``namespace/name``.
Based on docs/specification.md and ADR-004 (Pydantic Validation).
"""
from __future__ import annotations
import re
from collections import OrderedDict
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator
from cleveragents.domain.models.core.tool import (
ResourceSlot,
ToolCapability,
ToolSource,
)
# ---------------------------------------------------------------------------
# Regex patterns
# ---------------------------------------------------------------------------
_SKILL_NAME_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$")
# ---------------------------------------------------------------------------
# Supporting models
# ---------------------------------------------------------------------------
class SkillToolRef(BaseModel):
"""Reference to a named tool in the Tool Registry.
The name is a namespaced tool name (``namespace/short_name``).
"""
name: str = Field(..., min_length=1, description="Namespaced tool name reference")
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class SkillInclude(BaseModel):
"""Reference to another skill to include (recursive composition).
When a skill includes another skill, the included skill's tools are
merged into the parent's flattened tool set.
"""
name: str = Field(..., min_length=1, description="Namespaced skill name to include")
overrides: dict[str, Any] | None = Field(
default=None,
description="Per-tool metadata overrides applied during inclusion",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class SkillInlineTool(BaseModel):
"""An anonymous inline tool definition embedded in a skill.
Inline tools do not exist in the Tool Registry -- they are defined
directly in the skill YAML and only live within the skill's scope.
"""
description: str = Field(
..., min_length=1, description="Description of the inline tool"
)
source: ToolSource = Field(..., description="Source type for the inline tool")
code: str | None = Field(
default=None, description="Inline code (required for source=custom)"
)
input_schema: dict[str, Any] | None = Field(
default=None, description="JSON Schema for tool inputs"
)
output_schema: dict[str, Any] | None = Field(
default=None, description="JSON Schema for tool outputs"
)
capability: ToolCapability | None = Field(
default=None, description="Capability metadata for the inline tool"
)
resource_slots: list[ResourceSlot] = Field(
default_factory=list, description="Resource slot declarations"
)
timeout: int = Field(300, ge=1, description="Execution timeout in seconds")
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class SkillMcpSource(BaseModel):
"""MCP server source for a skill.
When ``tools`` is None, all tools from the server are imported.
"""
server: str = Field(..., min_length=1, description="MCP server command or URI")
tools: list[str] | None = Field(
default=None,
description="Specific tools to import (None means all)",
)
env: dict[str, str] | None = Field(
default=None, description="Environment variables for the MCP server"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class SkillAgentSource(BaseModel):
"""Agent Skills Standard folder source for a skill."""
path: str = Field(
..., min_length=1, description="Path to Agent Skills Standard folder"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
class SkillCapabilitySummary(BaseModel):
"""Aggregated capability summary for all tools in a resolved skill.
Computed after flattening and used for display and safety decisions.
"""
total_tools: int = Field(0, ge=0, description="Total number of tools")
read_only_tools: int = Field(0, ge=0, description="Number of read-only tools")
write_tools: int = Field(0, ge=0, description="Number of write-capable tools")
checkpointable_tools: int = Field(
0, ge=0, description="Number of checkpointable tools"
)
has_side_effects: bool = Field(
False, description="Whether any tool has side effects"
)
mcp_sources: int = Field(0, ge=0, description="Number of MCP server sources")
agent_skill_sources: int = Field(
0, ge=0, description="Number of Agent Skill sources"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# Skill
# ---------------------------------------------------------------------------
class Skill(BaseModel):
"""Domain model for a Skill.
A Skill is a namespaced, reusable collection of tools. The **namespaced
name** (``namespace/short_name``) is the unique identifier.
Created via ``agents skill add --config <file>``.
"""
# Identity -- namespace/name format
name: str = Field(
...,
min_length=1,
description="Namespaced skill name (namespace/short_name)",
)
description: str = Field(
...,
min_length=1,
description="Short description of what the skill provides",
)
# Tool references -- named tools from the Tool Registry
tool_refs: list[str] = Field(
default_factory=list,
description="Named tool references (namespace/name)",
)
# Included skills -- recursive composition
includes: list[SkillInclude] = Field(
default_factory=list,
description="Other skills to include",
)
# Inline tool definitions
anonymous_tools: list[SkillInlineTool] = Field(
default_factory=list,
description="Anonymous inline tool definitions",
)
# External sources
mcp_servers: list[SkillMcpSource] = Field(
default_factory=list,
description="MCP server sources",
)
agent_skills: list[SkillAgentSource] = Field(
default_factory=list,
description="Agent Skills Standard folder sources",
)
# Per-tool metadata overrides
overrides: dict[str, dict[str, Any]] = Field(
default_factory=dict,
description="Per-tool metadata overrides keyed by tool name",
)
# -- Name format validation ------------------------------------------------
@field_validator("name")
@classmethod
def validate_name_format(cls: type[Skill], v: str) -> str:
"""Validate skill name matches namespace/name pattern."""
if not _SKILL_NAME_PATTERN.match(v):
raise ValueError(
f"Skill name must match 'namespace/name' pattern "
f"(alphanumeric, hyphens, underscores): got '{v}'"
)
return v
# -- Properties ------------------------------------------------------------
@property
def namespace(self) -> str:
"""Extract the namespace portion from the name."""
return self.name.split("/", 1)[0]
@property
def short_name(self) -> str:
"""Extract the short name portion from the name."""
return self.name.split("/", 1)[1]
# -- Factories & serialization --------------------------------------------
@classmethod
def from_config(cls, config: dict[str, Any]) -> Skill:
"""Create a Skill from a YAML configuration dict.
The config follows the spec schema (see ``docs/specification.md``
Skill Configuration Files section).
"""
if "name" not in config:
raise ValueError("Skill config must include 'name'")
if "description" not in config:
raise ValueError("Skill config must include 'description'")
# Build includes from list of dicts or strings
includes_raw: list[dict[str, Any] | str] = config.get("includes", [])
includes: list[SkillInclude] = []
for item in includes_raw:
if isinstance(item, str):
includes.append(SkillInclude(name=item))
else:
includes.append(SkillInclude.model_validate(item))
# Build anonymous tools
anon_raw: list[dict[str, Any]] = config.get("anonymous_tools", [])
anonymous_tools = [SkillInlineTool.model_validate(t) for t in anon_raw]
# Build MCP sources
mcp_raw: list[dict[str, Any]] = config.get("mcp_servers", [])
mcp_servers = [SkillMcpSource.model_validate(m) for m in mcp_raw]
# Build agent skill sources
agent_raw: list[dict[str, Any] | str] = config.get("agent_skills", [])
agent_skills: list[SkillAgentSource] = []
for item in agent_raw:
if isinstance(item, str):
agent_skills.append(SkillAgentSource(path=item))
else:
agent_skills.append(SkillAgentSource.model_validate(item))
return cls(
name=config["name"],
description=config["description"],
tool_refs=config.get("tools", config.get("tool_refs", [])),
includes=includes,
anonymous_tools=anonymous_tools,
mcp_servers=mcp_servers,
agent_skills=agent_skills,
overrides=config.get("overrides", {}),
)
def as_cli_dict(self) -> OrderedDict[str, Any]:
"""Return a stable-ordered dictionary for CLI rendering.
Keys are ordered for consistent display across output formats.
"""
result: OrderedDict[str, Any] = OrderedDict()
result["name"] = self.name
result["description"] = self.description
result["namespace"] = self.namespace
result["short_name"] = self.short_name
if self.tool_refs:
result["tool_refs"] = list(self.tool_refs)
if self.includes:
result["includes"] = [
{
"name": inc.name,
**({"overrides": inc.overrides} if inc.overrides else {}),
}
for inc in self.includes
]
if self.anonymous_tools:
result["anonymous_tools"] = [
{
"description": t.description,
"source": t.source.value,
**({"code": t.code} if t.code else {}),
"timeout": t.timeout,
}
for t in self.anonymous_tools
]
if self.mcp_servers:
result["mcp_servers"] = [
{
"server": m.server,
**({"tools": m.tools} if m.tools is not None else {}),
**({"env": m.env} if m.env else {}),
}
for m in self.mcp_servers
]
if self.agent_skills:
result["agent_skills"] = [{"path": a.path} for a in self.agent_skills]
if self.overrides:
result["overrides"] = dict(self.overrides)
return result
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
use_enum_values=False,
)
# ---------------------------------------------------------------------------
# Resolved tool entry (output of SkillResolver)
# ---------------------------------------------------------------------------
class ResolvedToolEntry(BaseModel):
"""A single tool entry in the resolved (flattened) tool set.
Tracks origin for debugging and override application.
"""
name: str = Field(..., description="Tool name (or generated key for inline)")
source_skill: str = Field(..., description="Skill that contributed this tool")
is_inline: bool = Field(False, description="Whether this is an inline tool")
inline_tool: SkillInlineTool | None = Field(
default=None, description="Inline tool definition (if is_inline)"
)
overrides: dict[str, Any] = Field(
default_factory=dict, description="Applied overrides"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# SkillResolver
# ---------------------------------------------------------------------------
class SkillResolver:
"""Resolves a skill into its flattened set of tools.
Handles recursive include resolution with cycle detection,
tool de-duplication (last-wins for overrides), and deterministic
ordering of the output.
"""
def resolve_tools(
self,
skill: Skill,
skill_lookup: dict[str, Skill],
) -> list[ResolvedToolEntry]:
"""Resolve the flattened tool set for a skill.
Args:
skill: The root skill to resolve.
skill_lookup: Map of skill names to Skill objects for include
resolution.
Returns:
Ordered list of resolved tool entries.
Raises:
ValueError: If a cycle is detected in skill includes.
"""
seen_tools: dict[str, ResolvedToolEntry] = {}
order: list[str] = []
self._resolve_recursive(
skill=skill,
skill_lookup=skill_lookup,
seen_tools=seen_tools,
order=order,
visited=set(),
path=[],
)
return [seen_tools[name] for name in order]
def _resolve_recursive(
self,
skill: Skill,
skill_lookup: dict[str, Skill],
seen_tools: dict[str, ResolvedToolEntry],
order: list[str],
visited: set[str],
path: list[str],
) -> None:
"""Recursively resolve includes, collecting tools along the way."""
if skill.name in visited:
cycle_path = " -> ".join([*path, skill.name])
raise ValueError(f"Cycle detected in skill includes: {cycle_path}")
visited.add(skill.name)
path.append(skill.name)
# 1. Resolve includes first (depth-first)
for include in skill.includes:
included_skill = skill_lookup.get(include.name)
if included_skill is None:
raise ValueError(
f"Included skill '{include.name}' not found "
f"(referenced from '{skill.name}')"
)
self._resolve_recursive(
skill=included_skill,
skill_lookup=skill_lookup,
seen_tools=seen_tools,
order=order,
visited=set(visited),
path=list(path),
)
# 2. Add tool_refs from this skill
for ref in skill.tool_refs:
entry = ResolvedToolEntry(
name=ref,
source_skill=skill.name,
is_inline=False,
)
# Apply overrides from the include that pulled this skill
if ref in skill.overrides:
entry = entry.model_copy(
update={"overrides": dict(skill.overrides[ref])}
)
if ref not in seen_tools:
order.append(ref)
seen_tools[ref] = entry
# 3. Add anonymous (inline) tools
for idx, inline in enumerate(skill.anonymous_tools):
key = f"{skill.name}/_anon_{idx}"
entry = ResolvedToolEntry(
name=key,
source_skill=skill.name,
is_inline=True,
inline_tool=inline,
)
if key not in seen_tools:
order.append(key)
seen_tools[key] = entry
# 4. Add MCP server tools (as references)
for mcp in skill.mcp_servers:
if mcp.tools:
for tool_name in mcp.tools:
mcp_key = f"mcp:{mcp.server}/{tool_name}"
entry = ResolvedToolEntry(
name=mcp_key,
source_skill=skill.name,
is_inline=False,
)
if mcp_key not in seen_tools:
order.append(mcp_key)
seen_tools[mcp_key] = entry
# 5. Add agent skill sources
for agent in skill.agent_skills:
agent_key = f"agent_skill:{agent.path}"
entry = ResolvedToolEntry(
name=agent_key,
source_skill=skill.name,
is_inline=False,
)
if agent_key not in seen_tools:
order.append(agent_key)
seen_tools[agent_key] = entry
path.pop()
def compute_capability_summary(
self,
skill: Skill,
resolved: list[ResolvedToolEntry],
) -> SkillCapabilitySummary:
"""Compute a capability summary from resolved tool entries.
Args:
skill: The original skill (for MCP/agent counts).
resolved: The resolved tool entries.
Returns:
Aggregated capability summary.
"""
read_only_count = 0
write_count = 0
checkpoint_count = 0
has_side_effects = False
for entry in resolved:
if entry.is_inline and entry.inline_tool is not None:
cap = entry.inline_tool.capability
if cap is not None:
if cap.read_only:
read_only_count += 1
if cap.writes:
write_count += 1
if cap.checkpointable:
checkpoint_count += 1
if cap.side_effects:
has_side_effects = True
return SkillCapabilitySummary(
total_tools=len(resolved),
read_only_tools=read_only_count,
write_tools=write_count,
checkpointable_tools=checkpoint_count,
has_side_effects=has_side_effects,
mcp_sources=len(skill.mcp_servers),
agent_skill_sources=len(skill.agent_skills),
)