feat(skill): add skill protocol and metadata

This commit is contained in:
2026-02-20 03:59:51 +00:00
parent 3bd02a7c6e
commit b18bace06b
9 changed files with 1957 additions and 24 deletions
+192
View File
@@ -0,0 +1,192 @@
"""ASV benchmarks for Skill protocol type validation throughput.
Measures the performance of:
- SkillMetadata.from_skill() factory
- SkillError creation
- SkillResult creation
- SkillDefinition creation with schema validation
- map_tool_error() error mapping
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.domain.models.core.skill import (
Skill,
SkillInlineTool,
SkillResolver,
)
from cleveragents.domain.models.core.tool import (
ToolCapability,
ToolSource,
)
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillErrorType,
SkillMetadata,
SkillResult,
map_tool_error,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.skill import (
Skill,
SkillInlineTool,
SkillResolver,
)
from cleveragents.domain.models.core.tool import (
ToolCapability,
ToolSource,
)
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillErrorType,
SkillMetadata,
SkillResult,
map_tool_error,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
_BASIC_SKILL = Skill(
name="local/bench-skill",
description="Benchmark skill",
tool_refs=["local/tool-a", "local/tool-b"],
)
_INLINE_SKILL = Skill(
name="local/bench-inline",
description="Benchmark inline skill",
anonymous_tools=[
SkillInlineTool(
description="Inline bench tool",
source=ToolSource.CUSTOM,
code="x = 1",
capability=ToolCapability(read_only=True),
),
],
)
_RESOLVER = SkillResolver()
_RESOLVED_BASIC = _RESOLVER.resolve_tools(_BASIC_SKILL, {})
_RESOLVED_INLINE = _RESOLVER.resolve_tools(_INLINE_SKILL, {})
_METADATA_BASIC = SkillMetadata.from_skill(_BASIC_SKILL, resolved=_RESOLVED_BASIC)
# ---------------------------------------------------------------------------
# Benchmarks
# ---------------------------------------------------------------------------
class TimeSkillMetadataFromSkill:
"""Benchmark SkillMetadata.from_skill() throughput."""
timeout = 60
def time_from_skill_basic(self) -> None:
"""Time metadata creation from a basic skill."""
SkillMetadata.from_skill(_BASIC_SKILL, resolved=_RESOLVED_BASIC)
def time_from_skill_inline(self) -> None:
"""Time metadata creation from a skill with inline tools."""
SkillMetadata.from_skill(_INLINE_SKILL, resolved=_RESOLVED_INLINE)
class TimeSkillErrorCreation:
"""Benchmark SkillError creation throughput."""
timeout = 60
def time_create_error(self) -> None:
"""Time SkillError instantiation."""
SkillError(
error_type=SkillErrorType.TOOL_EXECUTION_FAILURE,
message="benchmark error",
skill_name="local/bench-skill",
details={"trace": "abc"},
)
class TimeSkillResultCreation:
"""Benchmark SkillResult creation throughput."""
timeout = 60
def time_create_success_result(self) -> None:
"""Time successful SkillResult instantiation."""
SkillResult(
skill_name="local/bench-skill",
tool_name="local/bench-tool",
success=True,
output_data={"value": 42},
duration_ms=10.5,
)
def time_create_failure_result(self) -> None:
"""Time failed SkillResult instantiation."""
err = SkillError(
error_type=SkillErrorType.TOOL_EXECUTION_FAILURE,
message="benchmark error",
skill_name="local/bench-skill",
)
SkillResult(
skill_name="local/bench-skill",
tool_name="local/bench-tool",
success=False,
error=err,
)
class TimeSkillDefinitionCreation:
"""Benchmark SkillDefinition creation throughput."""
timeout = 60
def time_create_definition(self) -> None:
"""Time SkillDefinition instantiation."""
SkillDefinition(
skill=_BASIC_SKILL,
resolved_tools=_RESOLVED_BASIC,
metadata=_METADATA_BASIC,
)
def time_create_definition_with_schemas(self) -> None:
"""Time SkillDefinition with input/output schemas."""
SkillDefinition(
skill=_BASIC_SKILL,
resolved_tools=_RESOLVED_BASIC,
metadata=_METADATA_BASIC,
input_schema={"type": "object", "properties": {"q": {"type": "string"}}},
output_schema={"type": "object", "properties": {"r": {"type": "string"}}},
)
class TimeMapToolError:
"""Benchmark map_tool_error() throughput."""
timeout = 60
def time_map_value_error(self) -> None:
"""Time mapping a ValueError."""
map_tool_error(ValueError("not found"), skill_name="local/bench-skill")
def time_map_permission_error(self) -> None:
"""Time mapping a PermissionError."""
map_tool_error(PermissionError("denied"), skill_name="local/bench-skill")
def time_map_runtime_error(self) -> None:
"""Time mapping a generic RuntimeError."""
map_tool_error(
RuntimeError("crash"),
skill_name="local/bench-skill",
tool_name="local/bench-tool",
)
+182
View File
@@ -0,0 +1,182 @@
# Skill Protocol Reference
The **Skill Protocol** defines the type contracts used for skill discovery,
composition, execution, and error reporting within CleverAgents v3.
All protocol types are frozen Pydantic `BaseModel` instances, ensuring
immutability after construction.
---
## Overview
| Type | Purpose |
|------|---------|
| `SkillErrorType` | `StrEnum` of all skill error categories |
| `SkillError` | Structured error payload |
| `SkillMetadata` | Lightweight discovery descriptor |
| `SkillDefinition` | Full skill + resolved tools + metadata |
| `SkillResult` | Outcome of a skill-tool invocation |
---
## SkillErrorType
A `StrEnum` with the following members:
| Value | Description |
|-------|-------------|
| `skill_not_found` | The requested skill does not exist |
| `resolution_failure` | Skill resolution failed (missing include, etc.) |
| `cycle_detected` | Circular dependency in skill includes |
| `tool_activation_failure` | Tool could not be activated |
| `tool_execution_failure` | Tool execution failed at runtime |
| `validation_error` | Input/output schema validation failed |
| `permission_denied` | Operation not permitted |
---
## SkillError
Structured error payload for skill operations.
### Fields
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| `error_type` | `SkillErrorType` | Yes | Error category |
| `message` | `str` | Yes | Human-readable description |
| `details` | `dict[str, Any]` | No | Diagnostic context |
| `skill_name` | `str` | Yes | Originating skill |
| `tool_name` | `str \| None` | No | Originating tool |
---
## SkillMetadata
Lightweight metadata for skill discovery and catalogue listings.
### Fields
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `name` | `str` | — | Namespaced skill name |
| `description` | `str` | — | Skill description |
| `version` | `str` | `"0.0.0"` | Semantic version |
| `tool_count` | `int` | `0` | Number of resolved tools |
| `capability_summary` | `SkillCapabilitySummary` | empty | Aggregated capabilities |
| `source_types` | `list[str]` | `[]` | Distinct source type labels |
| `writes` | `bool` | `False` | Any tool can write |
| `read_only` | `bool` | `True` | Entire skill is read-only |
### Factory: `from_skill()`
```python
meta = SkillMetadata.from_skill(skill, resolved=resolved_entries, version="1.0.0")
```
Derives `writes` and `read_only` from the `SkillCapabilitySummary`:
- `writes = True` when `write_tools > 0` or `has_side_effects` is true
- `read_only = True` when `write_tools == 0` and no side effects
### Writes / Read-Only Gating
The `writes` and `read_only` flags are propagated to the ToolRuntime
gating layer. A skill marked `read_only=True` will never be allowed
to trigger a write-capable tool at runtime.
---
## SkillDefinition
Full skill definition combining the domain `Skill` model, resolved tools,
and pre-computed metadata.
### Fields
| Field | Type | Required | Description |
|-------|------|----------|-------------|
| `skill` | `Skill` | Yes | Domain-level Skill model |
| `resolved_tools` | `list[ResolvedToolEntry]` | No | Flattened tool set |
| `metadata` | `SkillMetadata` | Yes | Pre-computed metadata |
| `input_schema` | `dict \| None` | No | JSON Schema for inputs |
| `output_schema` | `dict \| None` | No | JSON Schema for outputs |
### JSON Schema Validation
When `input_schema` or `output_schema` is provided, it **must** include a
`"type"` key to be considered a valid JSON Schema object. The validator
rejects schemas missing this key at construction time.
```json
{
"type": "object",
"properties": {
"query": {"type": "string"}
}
}
```
### Writes Consistency
The model validator checks that if `metadata.read_only` is `True`, none
of the resolved inline tools declare `writes=True`. A mismatch raises
a `ValidationError`.
---
## SkillResult
Outcome of a single skill-tool invocation.
### Fields
| Field | Type | Default | Description |
|-------|------|---------|-------------|
| `skill_name` | `str` | — | Invoked skill |
| `tool_name` | `str` | — | Invoked tool |
| `success` | `bool` | — | Whether invocation succeeded |
| `output_data` | `Any` | `None` | Tool output |
| `error` | `SkillError \| None` | `None` | Error on failure |
| `duration_ms` | `float` | `0.0` | Execution time in ms |
| `resource_changes` | `list[dict]` | `[]` | Resource change descriptors |
---
## Error Mapping
The `map_tool_error()` helper normalises arbitrary exceptions into
`SkillError` payloads:
```python
from cleveragents.skills.protocol import map_tool_error
try:
run_tool(...)
except Exception as exc:
error = map_tool_error(exc, skill_name="local/my-skill", tool_name="local/my-tool")
```
### Mapping Rules
| Exception Type | Message Contains | Mapped To |
|----------------|------------------|-----------|
| `ValueError` | `"cycle"` | `CYCLE_DETECTED` |
| `ValueError` | `"not found"` | `SKILL_NOT_FOUND` |
| `ValueError` | (other) | `RESOLUTION_FAILURE` |
| `PermissionError` | — | `PERMISSION_DENIED` |
| `ValidationError` | — | `VALIDATION_ERROR` |
| Any | `"activation"` | `TOOL_ACTIVATION_FAILURE` |
| Any | (other) | `TOOL_EXECUTION_FAILURE` |
---
## JSON Schema Rules Summary
1. **Input/output schemas** on `SkillDefinition` must include `"type"`.
2. **Writes/read_only metadata** on `SkillMetadata` must be consistent
with the resolved tool capabilities.
3. **Error payloads** always include `skill_name`; `tool_name` is
optional but recommended.
4. All protocol types are **frozen** (immutable after construction).
+201
View File
@@ -0,0 +1,201 @@
Feature: Skill Protocol Types
As a developer
I want skill protocol types for metadata, definition, result, and error
So that skills can be discovered, composed, executed, and errors reported uniformly
# ---- SkillErrorType Enum ----
Scenario: SkillErrorType has all expected members
When I list all skill_protocol error types
Then the skill_protocol error types should include "skill_not_found"
And the skill_protocol error types should include "resolution_failure"
And the skill_protocol error types should include "cycle_detected"
And the skill_protocol error types should include "tool_activation_failure"
And the skill_protocol error types should include "tool_execution_failure"
And the skill_protocol error types should include "validation_error"
And the skill_protocol error types should include "permission_denied"
Scenario: SkillErrorType has exactly 7 members
When I list all skill_protocol error types
Then there should be 7 skill_protocol error types
# ---- SkillError Creation ----
Scenario: Create a SkillError with required fields
When I create a skill_protocol error with type "tool_execution_failure" and message "Tool timed out" for skill "local/my-skill"
Then the skill_protocol error should be created
And the skill_protocol error type should be "tool_execution_failure"
And the skill_protocol error message should be "Tool timed out"
And the skill_protocol error skill_name should be "local/my-skill"
And the skill_protocol error tool_name should be none
Scenario: Create a SkillError with optional tool_name
When I create a skill_protocol error with tool_name "local/my-tool"
Then the skill_protocol error should be created
And the skill_protocol error tool_name should be "local/my-tool"
Scenario: Create a SkillError with details
When I create a skill_protocol error with details
Then the skill_protocol error should be created
And the skill_protocol error details should have key "trace_id"
Scenario: SkillError is frozen (immutable)
When I create a skill_protocol error with type "validation_error" and message "Bad input" for skill "local/test"
Then the skill_protocol error should be immutable
# ---- SkillMetadata Creation ----
Scenario: Create SkillMetadata from Skill domain model
When I create a skill_protocol metadata from a basic skill
Then the skill_protocol metadata should be created
And the skill_protocol metadata name should be "local/test-skill"
And the skill_protocol metadata description should be "A test skill"
And the skill_protocol metadata version should be "0.0.0"
And the skill_protocol metadata read_only should be true
And the skill_protocol metadata writes should be false
Scenario: Create SkillMetadata from Skill with tool refs
When I create a skill_protocol metadata from a skill with tool refs
Then the skill_protocol metadata should be created
And the skill_protocol metadata tool_count should be 2
And the skill_protocol metadata source_types should contain "tool_ref"
Scenario: Create SkillMetadata from Skill with inline write tools
When I create a skill_protocol metadata from a skill with write tools
Then the skill_protocol metadata should be created
And the skill_protocol metadata writes should be true
And the skill_protocol metadata read_only should be false
Scenario: Create SkillMetadata from Skill with MCP sources
When I create a skill_protocol metadata from a skill with mcp sources
Then the skill_protocol metadata should be created
And the skill_protocol metadata source_types should contain "mcp"
Scenario: Create SkillMetadata from Skill with agent skill sources
When I create a skill_protocol metadata from a skill with agent skill sources
Then the skill_protocol metadata should be created
And the skill_protocol metadata source_types should contain "agent_skill"
Scenario: Create SkillMetadata with custom version
When I create a skill_protocol metadata with version "1.2.3"
Then the skill_protocol metadata should be created
And the skill_protocol metadata version should be "1.2.3"
Scenario: SkillMetadata is frozen (immutable)
When I create a skill_protocol metadata from a basic skill
Then the skill_protocol metadata should be immutable
Scenario: SkillMetadata from_skill rejects None
When I try to create skill_protocol metadata from None
Then a skill_protocol value error should be raised
# ---- SkillDefinition ----
Scenario: Create a SkillDefinition with resolved tools
When I create a skill_protocol definition with resolved tools
Then the skill_protocol definition should be created
And the skill_protocol definition skill name should be "local/def-skill"
And the skill_protocol definition should have 1 resolved tools
Scenario: Create a SkillDefinition with input and output schemas
When I create a skill_protocol definition with input and output schemas
Then the skill_protocol definition should be created
And the skill_protocol definition input_schema should have key "type"
And the skill_protocol definition output_schema should have key "type"
Scenario: SkillDefinition rejects input_schema without type key
When I try to create a skill_protocol definition with invalid input_schema
Then a skill_protocol validation error should be raised
Scenario: SkillDefinition rejects output_schema without type key
When I try to create a skill_protocol definition with invalid output_schema
Then a skill_protocol validation error should be raised
Scenario: SkillDefinition detects writes inconsistency
When I try to create a skill_protocol definition with writes inconsistency
Then a skill_protocol validation error should be raised
Scenario: SkillDefinition is frozen (immutable)
When I create a skill_protocol definition with resolved tools
Then the skill_protocol definition should be immutable
# ---- SkillResult ----
Scenario: Create a successful SkillResult
When I create a skill_protocol result with success
Then the skill_protocol result should be created
And the skill_protocol result success should be true
And the skill_protocol result skill_name should be "local/my-skill"
And the skill_protocol result tool_name should be "local/my-tool"
And the skill_protocol result error should be none
And the skill_protocol result output_data should be "hello"
Scenario: Create a failed SkillResult with error
When I create a skill_protocol result with failure
Then the skill_protocol result should be created
And the skill_protocol result success should be false
And the skill_protocol result error should not be none
And the skill_protocol result error type should be "tool_execution_failure"
Scenario: Create a SkillResult with duration and resource changes
When I create a skill_protocol result with duration and resource changes
Then the skill_protocol result should be created
And the skill_protocol result duration_ms should be 150.5
And the skill_protocol result should have 1 resource changes
Scenario: SkillResult is frozen (immutable)
When I create a skill_protocol result with success
Then the skill_protocol result should be immutable
# ---- Error Mapping ----
Scenario: Map ValueError with cycle message to CYCLE_DETECTED
When I map a skill_protocol ValueError with message "Cycle detected in skill includes"
Then the skill_protocol mapped error type should be "cycle_detected"
Scenario: Map ValueError with not found message to SKILL_NOT_FOUND
When I map a skill_protocol ValueError with message "Skill not found"
Then the skill_protocol mapped error type should be "skill_not_found"
Scenario: Map generic ValueError to RESOLUTION_FAILURE
When I map a skill_protocol ValueError with message "Something went wrong"
Then the skill_protocol mapped error type should be "resolution_failure"
Scenario: Map PermissionError to PERMISSION_DENIED
When I map a skill_protocol PermissionError
Then the skill_protocol mapped error type should be "permission_denied"
Scenario: Map RuntimeError to TOOL_EXECUTION_FAILURE
When I map a skill_protocol RuntimeError with message "Unexpected failure"
Then the skill_protocol mapped error type should be "tool_execution_failure"
Scenario: Map RuntimeError with activation in message
When I map a skill_protocol RuntimeError with message "activation failed"
Then the skill_protocol mapped error type should be "tool_activation_failure"
Scenario: Error mapping with tool_name populated
When I map a skill_protocol error with tool_name "local/my-tool"
Then the skill_protocol mapped error tool_name should be "local/my-tool"
And the skill_protocol mapped error details should have key "tool_name"
Scenario: Error mapping rejects empty skill_name
When I try to map a skill_protocol error with empty skill_name
Then a skill_protocol value error should be raised
# ---- Writes/Read-Only Metadata Propagation ----
Scenario: Skill with only read-only tools has read_only=True
When I create a skill_protocol metadata from a skill with read_only tools
Then the skill_protocol metadata read_only should be true
And the skill_protocol metadata writes should be false
Scenario: Skill with write tools has writes=True
When I create a skill_protocol metadata from a skill with write tools
Then the skill_protocol metadata writes should be true
And the skill_protocol metadata read_only should be false
Scenario: Empty skill has read_only=True
When I create a skill_protocol metadata from an empty skill
Then the skill_protocol metadata read_only should be true
And the skill_protocol metadata writes should be false
And the skill_protocol metadata tool_count should be 0
+709
View File
@@ -0,0 +1,709 @@
"""Step definitions for Skill protocol types tests."""
from typing import Any
from behave import then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.skill import (
Skill,
SkillAgentSource,
SkillInlineTool,
SkillMcpSource,
SkillResolver,
)
from cleveragents.domain.models.core.tool import (
ToolCapability,
ToolSource,
)
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillErrorType,
SkillMetadata,
SkillResult,
map_tool_error,
)
# ---------------------------------------------------------------------------
# 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 _make_metadata_skill_with_write_tools() -> Skill:
"""Create a skill with inline write-capable tools."""
return Skill(
name="local/write-skill",
description="Skill with writes",
anonymous_tools=[
SkillInlineTool(
description="Write tool",
source=ToolSource.CUSTOM,
code="x = 1",
capability=ToolCapability(
writes=True,
side_effects=["filesystem"],
),
),
],
)
def _make_metadata_skill_with_read_only_tools() -> Skill:
"""Create a skill with only read-only inline tools."""
return Skill(
name="local/readonly-skill",
description="Read only skill",
anonymous_tools=[
SkillInlineTool(
description="Read tool",
source=ToolSource.CUSTOM,
code="x = 1",
capability=ToolCapability(read_only=True),
),
],
)
# ---------------------------------------------------------------------------
# SkillErrorType Enum
# ---------------------------------------------------------------------------
@when("I list all skill_protocol error types")
def skill_protocol_list_error_types(context: Context) -> None:
"""Collect all SkillErrorType members."""
context.skill_protocol_error_types = list(SkillErrorType)
@then('the skill_protocol error types should include "{value}"')
def skill_protocol_check_error_type_member(context: Context, value: str) -> None:
"""Check that a specific error type is in the list."""
values = [e.value for e in context.skill_protocol_error_types]
assert value in values, f"Expected '{value}' in {values}"
@then("there should be {count:d} skill_protocol error types")
def skill_protocol_check_error_type_count(context: Context, count: int) -> None:
"""Check total number of error types."""
actual = len(context.skill_protocol_error_types)
assert actual == count, f"Expected {count} error types, got {actual}"
# ---------------------------------------------------------------------------
# SkillError Creation
# ---------------------------------------------------------------------------
@when(
'I create a skill_protocol error with type "{err_type}" and message "{msg}" for skill "{skill_name}"'
)
def skill_protocol_create_error(
context: Context, err_type: str, msg: str, skill_name: str
) -> None:
"""Create a SkillError."""
context.skill_protocol_error = SkillError(
error_type=SkillErrorType(err_type),
message=msg,
skill_name=skill_name,
)
@when('I create a skill_protocol error with tool_name "{tool_name}"')
def skill_protocol_create_error_with_tool(context: Context, tool_name: str) -> None:
"""Create a SkillError with tool_name."""
context.skill_protocol_error = SkillError(
error_type=SkillErrorType.TOOL_EXECUTION_FAILURE,
message="Tool failed",
skill_name="local/test-skill",
tool_name=tool_name,
)
@when("I create a skill_protocol error with details")
def skill_protocol_create_error_with_details(context: Context) -> None:
"""Create a SkillError with details dict."""
context.skill_protocol_error = SkillError(
error_type=SkillErrorType.TOOL_EXECUTION_FAILURE,
message="Tool failed",
skill_name="local/test-skill",
details={"trace_id": "abc-123", "attempt": 1},
)
@then("the skill_protocol error should be created")
def skill_protocol_check_error_created(context: Context) -> None:
"""Verify error was created."""
assert context.skill_protocol_error is not None
@then('the skill_protocol error type should be "{expected}"')
def skill_protocol_check_error_type(context: Context, expected: str) -> None:
"""Check error type value."""
assert context.skill_protocol_error.error_type.value == expected
@then('the skill_protocol error message should be "{expected}"')
def skill_protocol_check_error_message(context: Context, expected: str) -> None:
"""Check error message."""
assert context.skill_protocol_error.message == expected
@then('the skill_protocol error skill_name should be "{expected}"')
def skill_protocol_check_error_skill_name(context: Context, expected: str) -> None:
"""Check error skill_name."""
assert context.skill_protocol_error.skill_name == expected
@then("the skill_protocol error tool_name should be none")
def skill_protocol_check_error_tool_none(context: Context) -> None:
"""Check error tool_name is None."""
assert context.skill_protocol_error.tool_name is None
@then('the skill_protocol error tool_name should be "{expected}"')
def skill_protocol_check_error_tool_name(context: Context, expected: str) -> None:
"""Check error tool_name value."""
assert context.skill_protocol_error.tool_name == expected
@then('the skill_protocol error details should have key "{key}"')
def skill_protocol_check_error_details_key(context: Context, key: str) -> None:
"""Check error details contains a key."""
assert key in context.skill_protocol_error.details
@then("the skill_protocol error should be immutable")
def skill_protocol_check_error_immutable(context: Context) -> None:
"""Verify SkillError is frozen."""
try:
context.skill_protocol_error.message = "changed"
raise AssertionError("Expected frozen model to reject assignment")
except ValidationError:
pass
# ---------------------------------------------------------------------------
# SkillMetadata Creation
# ---------------------------------------------------------------------------
@when("I create a skill_protocol metadata from a basic skill")
def skill_protocol_create_metadata_basic(context: Context) -> None:
"""Create SkillMetadata from a basic skill."""
skill = _make_skill()
context.skill_protocol_metadata = SkillMetadata.from_skill(skill)
@when("I create a skill_protocol metadata from a skill with tool refs")
def skill_protocol_create_metadata_refs(context: Context) -> None:
"""Create SkillMetadata from a skill with tool refs."""
skill = _make_skill(tool_refs=["local/edit-file", "local/read-file"])
context.skill_protocol_metadata = SkillMetadata.from_skill(skill)
@when("I create a skill_protocol metadata from a skill with write tools")
def skill_protocol_create_metadata_write(context: Context) -> None:
"""Create SkillMetadata from a skill with write tools."""
skill = _make_metadata_skill_with_write_tools()
context.skill_protocol_metadata = SkillMetadata.from_skill(skill)
@when("I create a skill_protocol metadata from a skill with mcp sources")
def skill_protocol_create_metadata_mcp(context: Context) -> None:
"""Create SkillMetadata from a skill with MCP sources."""
skill = Skill(
name="local/mcp-skill",
description="MCP skill",
mcp_servers=[SkillMcpSource(server="npx @mcp/fs", tools=["read"])],
)
context.skill_protocol_metadata = SkillMetadata.from_skill(skill)
@when("I create a skill_protocol metadata from a skill with agent skill sources")
def skill_protocol_create_metadata_agent(context: Context) -> None:
"""Create SkillMetadata from a skill with agent skill sources."""
skill = Skill(
name="local/agent-skill",
description="Agent skill",
agent_skills=[SkillAgentSource(path="./skills/review")],
)
context.skill_protocol_metadata = SkillMetadata.from_skill(skill)
@when('I create a skill_protocol metadata with version "{version}"')
def skill_protocol_create_metadata_version(context: Context, version: str) -> None:
"""Create SkillMetadata with a specific version."""
skill = _make_skill()
context.skill_protocol_metadata = SkillMetadata.from_skill(skill, version=version)
@when("I create a skill_protocol metadata from a skill with read_only tools")
def skill_protocol_create_metadata_readonly(context: Context) -> None:
"""Create SkillMetadata from a skill with read-only tools."""
skill = _make_metadata_skill_with_read_only_tools()
context.skill_protocol_metadata = SkillMetadata.from_skill(skill)
@when("I create a skill_protocol metadata from an empty skill")
def skill_protocol_create_metadata_empty(context: Context) -> None:
"""Create SkillMetadata from an empty skill."""
skill = _make_skill()
context.skill_protocol_metadata = SkillMetadata.from_skill(skill)
@when("I try to create skill_protocol metadata from None")
def skill_protocol_create_metadata_none(context: Context) -> None:
"""Try to create SkillMetadata from None."""
context.skill_protocol_value_error = None
try:
SkillMetadata.from_skill(None) # type: ignore[arg-type]
except ValueError as e:
context.skill_protocol_value_error = e
@then("the skill_protocol metadata should be created")
def skill_protocol_check_metadata_created(context: Context) -> None:
"""Verify metadata was created."""
assert context.skill_protocol_metadata is not None
@then('the skill_protocol metadata name should be "{expected}"')
def skill_protocol_check_metadata_name(context: Context, expected: str) -> None:
"""Check metadata name."""
assert context.skill_protocol_metadata.name == expected
@then('the skill_protocol metadata description should be "{expected}"')
def skill_protocol_check_metadata_desc(context: Context, expected: str) -> None:
"""Check metadata description."""
assert context.skill_protocol_metadata.description == expected
@then('the skill_protocol metadata version should be "{expected}"')
def skill_protocol_check_metadata_version(context: Context, expected: str) -> None:
"""Check metadata version."""
assert context.skill_protocol_metadata.version == expected
@then("the skill_protocol metadata read_only should be true")
def skill_protocol_check_metadata_readonly_true(context: Context) -> None:
"""Check metadata read_only is True."""
assert context.skill_protocol_metadata.read_only is True
@then("the skill_protocol metadata read_only should be false")
def skill_protocol_check_metadata_readonly_false(context: Context) -> None:
"""Check metadata read_only is False."""
assert context.skill_protocol_metadata.read_only is False
@then("the skill_protocol metadata writes should be true")
def skill_protocol_check_metadata_writes_true(context: Context) -> None:
"""Check metadata writes is True."""
assert context.skill_protocol_metadata.writes is True
@then("the skill_protocol metadata writes should be false")
def skill_protocol_check_metadata_writes_false(context: Context) -> None:
"""Check metadata writes is False."""
assert context.skill_protocol_metadata.writes is False
@then("the skill_protocol metadata tool_count should be {expected:d}")
def skill_protocol_check_metadata_tool_count(context: Context, expected: int) -> None:
"""Check metadata tool_count."""
actual = context.skill_protocol_metadata.tool_count
assert actual == expected, f"Expected {expected} tools, got {actual}"
@then('the skill_protocol metadata source_types should contain "{expected}"')
def skill_protocol_check_metadata_source_type(context: Context, expected: str) -> None:
"""Check metadata source_types contains a value."""
assert expected in context.skill_protocol_metadata.source_types, (
f"Expected '{expected}' in {context.skill_protocol_metadata.source_types}"
)
@then("the skill_protocol metadata should be immutable")
def skill_protocol_check_metadata_immutable(context: Context) -> None:
"""Verify SkillMetadata is frozen."""
try:
context.skill_protocol_metadata.name = "changed"
raise AssertionError("Expected frozen model to reject assignment")
except ValidationError:
pass
# ---------------------------------------------------------------------------
# SkillDefinition
# ---------------------------------------------------------------------------
@when("I create a skill_protocol definition with resolved tools")
def skill_protocol_create_definition(context: Context) -> None:
"""Create a SkillDefinition with resolved tools."""
skill = Skill(
name="local/def-skill",
description="Definition skill",
tool_refs=["local/my-tool"],
)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
metadata = SkillMetadata.from_skill(skill, resolved=resolved)
context.skill_protocol_definition = SkillDefinition(
skill=skill,
resolved_tools=resolved,
metadata=metadata,
)
@when("I create a skill_protocol definition with input and output schemas")
def skill_protocol_create_definition_schemas(context: Context) -> None:
"""Create a SkillDefinition with schemas."""
skill = Skill(
name="local/schema-skill",
description="Schema skill",
tool_refs=["local/my-tool"],
)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
metadata = SkillMetadata.from_skill(skill, resolved=resolved)
context.skill_protocol_definition = SkillDefinition(
skill=skill,
resolved_tools=resolved,
metadata=metadata,
input_schema={"type": "object", "properties": {"query": {"type": "string"}}},
output_schema={"type": "object", "properties": {"result": {"type": "string"}}},
)
@when("I try to create a skill_protocol definition with invalid input_schema")
def skill_protocol_create_definition_invalid_input(context: Context) -> None:
"""Try to create a SkillDefinition with invalid input_schema."""
skill = _make_skill()
metadata = SkillMetadata.from_skill(skill)
context.skill_protocol_validation_error = None
try:
SkillDefinition(
skill=skill,
resolved_tools=[],
metadata=metadata,
input_schema={"properties": {"x": {"type": "string"}}}, # missing "type"
)
except ValidationError as e:
context.skill_protocol_validation_error = e
@when("I try to create a skill_protocol definition with invalid output_schema")
def skill_protocol_create_definition_invalid_output(context: Context) -> None:
"""Try to create a SkillDefinition with invalid output_schema."""
skill = _make_skill()
metadata = SkillMetadata.from_skill(skill)
context.skill_protocol_validation_error = None
try:
SkillDefinition(
skill=skill,
resolved_tools=[],
metadata=metadata,
output_schema={"description": "no type key"},
)
except ValidationError as e:
context.skill_protocol_validation_error = e
@when("I try to create a skill_protocol definition with writes inconsistency")
def skill_protocol_create_definition_writes_inconsistency(context: Context) -> None:
"""Try to create a SkillDefinition where metadata says read_only but tools have writes."""
skill = Skill(
name="local/inconsistent",
description="Inconsistent skill",
anonymous_tools=[
SkillInlineTool(
description="Writer",
source=ToolSource.CUSTOM,
code="x = 1",
capability=ToolCapability(writes=True, side_effects=["fs"]),
),
],
)
# Build resolved tools -- they include a write-capable inline tool
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
# Force metadata to claim read_only=True (bypassing from_skill)
metadata = SkillMetadata(
name=skill.name,
description=skill.description,
tool_count=len(resolved),
writes=False,
read_only=True,
)
context.skill_protocol_validation_error = None
try:
SkillDefinition(
skill=skill,
resolved_tools=resolved,
metadata=metadata,
)
except ValidationError as e:
context.skill_protocol_validation_error = e
@then("the skill_protocol definition should be created")
def skill_protocol_check_definition_created(context: Context) -> None:
"""Verify definition was created."""
assert context.skill_protocol_definition is not None
@then('the skill_protocol definition skill name should be "{expected}"')
def skill_protocol_check_definition_name(context: Context, expected: str) -> None:
"""Check definition skill name."""
assert context.skill_protocol_definition.skill.name == expected
@then("the skill_protocol definition should have {count:d} resolved tools")
def skill_protocol_check_definition_tool_count(context: Context, count: int) -> None:
"""Check definition resolved_tools count."""
actual = len(context.skill_protocol_definition.resolved_tools)
assert actual == count, f"Expected {count}, got {actual}"
@then('the skill_protocol definition input_schema should have key "{key}"')
def skill_protocol_check_definition_input_key(context: Context, key: str) -> None:
"""Check definition input_schema has key."""
assert context.skill_protocol_definition.input_schema is not None
assert key in context.skill_protocol_definition.input_schema
@then('the skill_protocol definition output_schema should have key "{key}"')
def skill_protocol_check_definition_output_key(context: Context, key: str) -> None:
"""Check definition output_schema has key."""
assert context.skill_protocol_definition.output_schema is not None
assert key in context.skill_protocol_definition.output_schema
@then("a skill_protocol validation error should be raised")
def skill_protocol_check_validation_error(context: Context) -> None:
"""Verify a validation error was raised."""
assert context.skill_protocol_validation_error is not None, (
"Expected a validation error"
)
@then("the skill_protocol definition should be immutable")
def skill_protocol_check_definition_immutable(context: Context) -> None:
"""Verify SkillDefinition is frozen."""
try:
context.skill_protocol_definition.input_schema = {"type": "object"}
raise AssertionError("Expected frozen model to reject assignment")
except ValidationError:
pass
# ---------------------------------------------------------------------------
# SkillResult
# ---------------------------------------------------------------------------
@when("I create a skill_protocol result with success")
def skill_protocol_create_result_success(context: Context) -> None:
"""Create a successful SkillResult."""
context.skill_protocol_result = SkillResult(
skill_name="local/my-skill",
tool_name="local/my-tool",
success=True,
output_data="hello",
)
@when("I create a skill_protocol result with failure")
def skill_protocol_create_result_failure(context: Context) -> None:
"""Create a failed SkillResult."""
err = SkillError(
error_type=SkillErrorType.TOOL_EXECUTION_FAILURE,
message="Tool timed out",
skill_name="local/my-skill",
tool_name="local/my-tool",
)
context.skill_protocol_result = SkillResult(
skill_name="local/my-skill",
tool_name="local/my-tool",
success=False,
error=err,
)
@when("I create a skill_protocol result with duration and resource changes")
def skill_protocol_create_result_duration(context: Context) -> None:
"""Create a SkillResult with duration and resource changes."""
context.skill_protocol_result = SkillResult(
skill_name="local/my-skill",
tool_name="local/my-tool",
success=True,
duration_ms=150.5,
resource_changes=[{"type": "file_modified", "path": "/tmp/test.txt"}],
)
@then("the skill_protocol result should be created")
def skill_protocol_check_result_created(context: Context) -> None:
"""Verify result was created."""
assert context.skill_protocol_result is not None
@then("the skill_protocol result success should be true")
def skill_protocol_check_result_success_true(context: Context) -> None:
"""Check result success is True."""
assert context.skill_protocol_result.success is True
@then("the skill_protocol result success should be false")
def skill_protocol_check_result_success_false(context: Context) -> None:
"""Check result success is False."""
assert context.skill_protocol_result.success is False
@then('the skill_protocol result skill_name should be "{expected}"')
def skill_protocol_check_result_skill(context: Context, expected: str) -> None:
"""Check result skill_name."""
assert context.skill_protocol_result.skill_name == expected
@then('the skill_protocol result tool_name should be "{expected}"')
def skill_protocol_check_result_tool(context: Context, expected: str) -> None:
"""Check result tool_name."""
assert context.skill_protocol_result.tool_name == expected
@then("the skill_protocol result error should be none")
def skill_protocol_check_result_error_none(context: Context) -> None:
"""Check result error is None."""
assert context.skill_protocol_result.error is None
@then("the skill_protocol result error should not be none")
def skill_protocol_check_result_error_present(context: Context) -> None:
"""Check result error is not None."""
assert context.skill_protocol_result.error is not None
@then('the skill_protocol result error type should be "{expected}"')
def skill_protocol_check_result_error_type(context: Context, expected: str) -> None:
"""Check result error type."""
assert context.skill_protocol_result.error is not None
assert context.skill_protocol_result.error.error_type.value == expected
@then('the skill_protocol result output_data should be "{expected}"')
def skill_protocol_check_result_output(context: Context, expected: str) -> None:
"""Check result output_data."""
assert context.skill_protocol_result.output_data == expected
@then("the skill_protocol result duration_ms should be {expected:g}")
def skill_protocol_check_result_duration(context: Context, expected: float) -> None:
"""Check result duration_ms."""
assert context.skill_protocol_result.duration_ms == expected
@then("the skill_protocol result should have {count:d} resource changes")
def skill_protocol_check_result_changes(context: Context, count: int) -> None:
"""Check result resource_changes count."""
actual = len(context.skill_protocol_result.resource_changes)
assert actual == count, f"Expected {count}, got {actual}"
@then("the skill_protocol result should be immutable")
def skill_protocol_check_result_immutable(context: Context) -> None:
"""Verify SkillResult is frozen."""
try:
context.skill_protocol_result.success = False
raise AssertionError("Expected frozen model to reject assignment")
except ValidationError:
pass
# ---------------------------------------------------------------------------
# Error Mapping
# ---------------------------------------------------------------------------
@when('I map a skill_protocol ValueError with message "{msg}"')
def skill_protocol_map_value_error(context: Context, msg: str) -> None:
"""Map a ValueError through map_tool_error."""
context.skill_protocol_mapped_error = map_tool_error(
ValueError(msg), skill_name="local/test-skill"
)
@when("I map a skill_protocol PermissionError")
def skill_protocol_map_permission_error(context: Context) -> None:
"""Map a PermissionError through map_tool_error."""
context.skill_protocol_mapped_error = map_tool_error(
PermissionError("Access denied"), skill_name="local/test-skill"
)
@when('I map a skill_protocol RuntimeError with message "{msg}"')
def skill_protocol_map_runtime_error(context: Context, msg: str) -> None:
"""Map a RuntimeError through map_tool_error."""
context.skill_protocol_mapped_error = map_tool_error(
RuntimeError(msg), skill_name="local/test-skill"
)
@when('I map a skill_protocol error with tool_name "{tool_name}"')
def skill_protocol_map_with_tool(context: Context, tool_name: str) -> None:
"""Map an error with tool_name."""
context.skill_protocol_mapped_error = map_tool_error(
RuntimeError("fail"),
skill_name="local/test-skill",
tool_name=tool_name,
)
@when("I try to map a skill_protocol error with empty skill_name")
def skill_protocol_map_empty_skill_name(context: Context) -> None:
"""Try to map an error with empty skill_name."""
context.skill_protocol_value_error = None
try:
map_tool_error(RuntimeError("fail"), skill_name="")
except ValueError as e:
context.skill_protocol_value_error = e
@then('the skill_protocol mapped error type should be "{expected}"')
def skill_protocol_check_mapped_type(context: Context, expected: str) -> None:
"""Check mapped error type."""
assert context.skill_protocol_mapped_error.error_type.value == expected
@then('the skill_protocol mapped error tool_name should be "{expected}"')
def skill_protocol_check_mapped_tool(context: Context, expected: str) -> None:
"""Check mapped error tool_name."""
assert context.skill_protocol_mapped_error.tool_name == expected
@then('the skill_protocol mapped error details should have key "{key}"')
def skill_protocol_check_mapped_details(context: Context, key: str) -> None:
"""Check mapped error details has key."""
assert key in context.skill_protocol_mapped_error.details
@then("a skill_protocol value error should be raised")
def skill_protocol_check_value_error(context: Context) -> None:
"""Verify a ValueError was raised."""
assert context.skill_protocol_value_error is not None, "Expected a ValueError"
+21 -21
View File
@@ -3407,27 +3407,27 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m3-actor-context` to `master` with description "Add actor context CLI commands and tests.".
**Parallel Group C3: Skill Protocol & Context [Jeff]** (critical path; depends on C1)
- [ ] **COMMIT (Owner: Jeff | Group: C3.protocol | Branch: feature/m3-skill-protocol | Planned: Day 16 | Expected: Day 20) - Commit message: "feat(skill): add skill protocol and metadata"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-skill-protocol`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master`
- [ ] Code [Jeff]: Define Skill protocol interface, SkillMetadata, SkillResult, and SkillError types in `src/cleveragents/skills/protocol.py`.
- [ ] Code [Jeff]: Add `SkillDefinition` model that references Tool Registry names and optional inline tool definitions.
- [ ] Code [Jeff]: Add error mapping helpers to normalize tool failures into SkillError payloads.
- [ ] Code [Jeff]: Require explicit `writes`/`read_only` metadata on skills and propagate to ToolRuntime gating.
- [ ] Code [Jeff]: Add schema validation for `inputs`/`outputs` in SkillDefinition to match Tool Registry schemas.
- [ ] Docs [Jeff]: Add `docs/reference/skills_protocol.md` describing metadata, tool composition, and JSON schema rules.
- [ ] Tests (Behave) [Jeff]: Add `features/skill_protocol.feature` for metadata validation and error capture.
- [ ] Tests (Robot) [Jeff]: Add `robot/skill_protocol.robot` smoke tests.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/skill_protocol_bench.py` for validation throughput.
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [ ] Git [Jeff]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [ ] Git [Jeff]: `git commit -m "feat(skill): add skill protocol and metadata"`
- [ ] Git [Jeff]: `git push -u origin feature/m3-skill-protocol`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m3-skill-protocol` to `master` with description "Add skill protocol, metadata models, and tests.".
- [ ] **COMMIT (Owner: Jeff | Group: C3.context | Branch: feature/m3-skill-protocol | Planned: Day 17 | Expected: Day 20) - Commit message: "feat(skill): add skill context and registry"**
- [X] **COMMIT (Owner: Jeff | Group: C3.protocol | Branch: feature/m3-skill-protocol | Planned: Day 16 | Expected: Day 23) - Commit message: "feat(skill): add skill protocol and metadata"**
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m3-skill-protocol`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master`
- [X] Code [Jeff]: Define Skill protocol interface, SkillMetadata, SkillResult, and SkillError types in `src/cleveragents/skills/protocol.py`.
- [X] Code [Jeff]: Add `SkillDefinition` model that references Tool Registry names and optional inline tool definitions.
- [X] Code [Jeff]: Add error mapping helpers to normalize tool failures into SkillError payloads.
- [X] Code [Jeff]: Require explicit `writes`/`read_only` metadata on skills and propagate to ToolRuntime gating.
- [X] Code [Jeff]: Add schema validation for `inputs`/`outputs` in SkillDefinition to match Tool Registry schemas.
- [X] Docs [Jeff]: Add `docs/reference/skills_protocol.md` describing metadata, tool composition, and JSON schema rules.
- [X] Tests (Behave) [Jeff]: Add `features/skill_protocol.feature` for metadata validation and error capture.
- [X] Tests (Robot) [Jeff]: Add `robot/skill_protocol.robot` smoke tests.
- [X] Tests (ASV) [Jeff]: Add `benchmarks/skill_protocol_bench.py` for validation throughput.
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [X] Git [Jeff]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [X] Git [Jeff]: `git commit -m "feat(skill): add skill protocol and metadata"`
- [X] Git [Jeff]: `git push -u origin feature/m3-skill-protocol`
- [X] Forgejo PR [Jeff]: Open PR from `feature/m3-skill-protocol` to `master` with description "Add skill protocol, metadata models, and tests.".
- [ ] **COMMIT (Owner: Jeff | Group: C3.context | Branch: feature/m3-skill-protocol | Planned: Day 17 | Expected: Day 24) - Commit message: "feat(skill): add skill context and registry"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-skill-protocol`
+153
View File
@@ -0,0 +1,153 @@
"""Robot Framework helper for skill protocol type validation.
Provides a CLI-style interface for Robot to invoke protocol type
creation and validation. Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_skill_protocol.py create-metadata
python robot/helper_skill_protocol.py create-error
python robot/helper_skill_protocol.py create-result
python robot/helper_skill_protocol.py create-definition
python robot/helper_skill_protocol.py map-error
python robot/helper_skill_protocol.py error-types
"""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.domain.models.core.skill import ( # noqa: E402
Skill,
SkillResolver,
)
from cleveragents.skills.protocol import ( # noqa: E402
SkillDefinition,
SkillError,
SkillErrorType,
SkillMetadata,
SkillResult,
map_tool_error,
)
def _cmd_create_metadata() -> int:
"""Test SkillMetadata.from_skill."""
skill = Skill(name="local/robot-skill", description="Robot test skill")
meta = SkillMetadata.from_skill(skill)
if meta.name != "local/robot-skill":
print(f"skill-protocol-fail: unexpected name {meta.name}")
return 1
print(f"skill-protocol-metadata-ok: {meta.name}")
return 0
def _cmd_create_error() -> int:
"""Test SkillError creation."""
err = SkillError(
error_type=SkillErrorType.TOOL_EXECUTION_FAILURE,
message="Test error",
skill_name="local/robot-skill",
)
if err.error_type != SkillErrorType.TOOL_EXECUTION_FAILURE:
print(f"skill-protocol-fail: unexpected type {err.error_type}")
return 1
print(f"skill-protocol-error-ok: {err.error_type.value}")
return 0
def _cmd_create_result() -> int:
"""Test SkillResult creation."""
result = SkillResult(
skill_name="local/robot-skill",
tool_name="local/robot-tool",
success=True,
output_data={"status": "ok"},
)
if not result.success:
print("skill-protocol-fail: result not success")
return 1
print("skill-protocol-result-ok")
return 0
def _cmd_create_definition() -> int:
"""Test SkillDefinition creation."""
skill = Skill(
name="local/robot-def",
description="Robot definition test",
tool_refs=["local/tool-a"],
)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
meta = SkillMetadata.from_skill(skill, resolved=resolved)
defn = SkillDefinition(
skill=skill,
resolved_tools=resolved,
metadata=meta,
)
if defn.skill.name != "local/robot-def":
print(f"skill-protocol-fail: unexpected name {defn.skill.name}")
return 1
print(f"skill-protocol-definition-ok: {defn.skill.name}")
return 0
def _cmd_map_error() -> int:
"""Test error mapping."""
mapped = map_tool_error(
ValueError("Cycle detected"),
skill_name="local/robot-skill",
)
if mapped.error_type != SkillErrorType.CYCLE_DETECTED:
print(f"skill-protocol-fail: unexpected type {mapped.error_type}")
return 1
print(f"skill-protocol-map-ok: {mapped.error_type.value}")
return 0
def _cmd_error_types() -> int:
"""Test SkillErrorType enum members."""
members = list(SkillErrorType)
if len(members) != 7:
print(f"skill-protocol-fail: expected 7 error types, got {len(members)}")
return 1
print(f"skill-protocol-error-types-ok: {len(members)}")
return 0
_COMMANDS = {
"create-metadata": _cmd_create_metadata,
"create-error": _cmd_create_error,
"create-result": _cmd_create_result,
"create-definition": _cmd_create_definition,
"map-error": _cmd_map_error,
"error-types": _cmd_error_types,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(
"Usage: helper_skill_protocol.py "
"<create-metadata|create-error|create-result|create-definition|map-error|error-types>"
)
return 1
command = sys.argv[1]
handler = _COMMANDS.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
return handler()
if __name__ == "__main__":
sys.exit(main())
+57
View File
@@ -0,0 +1,57 @@
*** Settings ***
Documentation Smoke tests for Skill protocol types
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_skill_protocol.py
*** Test Cases ***
Create Skill Metadata
[Documentation] Create SkillMetadata from a Skill domain model
${result}= Run Process ${PYTHON} ${HELPER} create-metadata cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-protocol-metadata-ok
Create Skill Error
[Documentation] Create a SkillError with required fields
${result}= Run Process ${PYTHON} ${HELPER} create-error cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-protocol-error-ok
Create Skill Result
[Documentation] Create a successful SkillResult
${result}= Run Process ${PYTHON} ${HELPER} create-result cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-protocol-result-ok
Create Skill Definition
[Documentation] Create a SkillDefinition with resolved tools
${result}= Run Process ${PYTHON} ${HELPER} create-definition cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-protocol-definition-ok
Map Tool Error
[Documentation] Map an exception to a SkillError via error mapping
${result}= Run Process ${PYTHON} ${HELPER} map-error cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-protocol-map-ok
Verify Error Types Count
[Documentation] SkillErrorType should have exactly 7 members
${result}= Run Process ${PYTHON} ${HELPER} error-types cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-protocol-error-types-ok
+22 -3
View File
@@ -1,15 +1,34 @@
"""Skill YAML configuration schema and validation.
"""Skill framework for CleverAgents v3.
This package provides the ``SkillConfigSchema`` Pydantic model for
loading, validating, and normalizing skill YAML configuration files.
loading, validating, and normalizing skill YAML configuration files,
as well as the skill protocol types used for discovery, composition,
execution, and error reporting within the skill framework.
Usage::
from cleveragents.skills.schema import SkillConfigSchema
from cleveragents.skills.protocol import SkillDefinition
config = SkillConfigSchema.from_yaml_file("examples/skills/single-tool.yaml")
"""
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillErrorType,
SkillMetadata,
SkillResult,
map_tool_error,
)
from cleveragents.skills.schema import SkillConfigSchema
__all__ = ["SkillConfigSchema"]
__all__ = [
"SkillConfigSchema",
"SkillDefinition",
"SkillError",
"SkillErrorType",
"SkillMetadata",
"SkillResult",
"map_tool_error",
]
+420
View File
@@ -0,0 +1,420 @@
"""Skill protocol types for CleverAgents v3.
Defines the protocol-level types for skill discovery, composition,
execution results, and structured errors. All models are frozen
Pydantic ``BaseModel`` instances following the project's immutable
domain-model conventions.
## Overview
- **SkillErrorType** -- ``StrEnum`` enumerating all skill error categories.
- **SkillError** -- Structured error payload with error type, message, and
optional diagnostic details.
- **SkillMetadata** -- Lightweight descriptor for discovery (name, version,
tool count, capability summary, writes/read_only flags).
- **SkillDefinition** -- Full skill definition pairing the domain ``Skill``
with its resolved tools and pre-computed metadata. Includes schema
validation for inputs/outputs.
- **SkillResult** -- Outcome of a single skill-tool invocation (success
flag, output data, timing, resource changes, optional error).
## Writes / Read-Only Gating
Every ``SkillMetadata`` exposes explicit ``writes`` and ``read_only`` flags
derived from the underlying ``SkillCapabilitySummary``. These flags are
intended for propagation to the ToolRuntime gating layer so that a
skill marked ``read_only=True`` never triggers a write-capable tool.
## Error Mapping
The ``map_tool_error`` helper normalises arbitrary tool-execution
exceptions into a ``SkillError`` with the appropriate ``SkillErrorType``,
ensuring uniform error payloads across all skill operations.
Based on ``docs/specification.md`` and ``implementation_plan.md`` task
C3.protocol.
"""
from __future__ import annotations
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, model_validator
from cleveragents.domain.models.core.skill import (
ResolvedToolEntry,
Skill,
SkillCapabilitySummary,
SkillResolver,
)
# ---------------------------------------------------------------------------
# SkillErrorType enum
# ---------------------------------------------------------------------------
class SkillErrorType(StrEnum):
"""Enumeration of all skill error categories.
Used as the discriminator in ``SkillError`` to enable typed error
handling across the skill framework.
"""
SKILL_NOT_FOUND = "skill_not_found"
RESOLUTION_FAILURE = "resolution_failure"
CYCLE_DETECTED = "cycle_detected"
TOOL_ACTIVATION_FAILURE = "tool_activation_failure"
TOOL_EXECUTION_FAILURE = "tool_execution_failure"
VALIDATION_ERROR = "validation_error"
PERMISSION_DENIED = "permission_denied"
# ---------------------------------------------------------------------------
# SkillError
# ---------------------------------------------------------------------------
class SkillError(BaseModel):
"""Structured error payload for skill operations.
Captures the error type, human-readable message, and optional
diagnostic context. Produced by error mapping helpers or raised
directly when a skill operation fails.
"""
error_type: SkillErrorType = Field(..., description="Category of the skill error")
message: str = Field(
..., min_length=1, description="Human-readable error description"
)
details: dict[str, Any] = Field(
default_factory=dict,
description="Additional diagnostic information",
)
skill_name: str = Field(
..., min_length=1, description="Skill that produced the error"
)
tool_name: str | None = Field(
default=None,
description="Tool that produced the error (if applicable)",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# SkillMetadata
# ---------------------------------------------------------------------------
class SkillMetadata(BaseModel):
"""Lightweight metadata descriptor for skill discovery.
Designed to be cheap to compute and transmit so that skill
catalogues and CLI listing commands can display essential
information without resolving the full tool set.
The ``writes`` and ``read_only`` flags are intended for propagation
to the ToolRuntime gating layer.
"""
name: str = Field(..., min_length=1, description="Namespaced skill name")
description: str = Field(..., min_length=1, description="Skill description")
version: str = Field(default="0.0.0", description="Semantic version of the skill")
tool_count: int = Field(
default=0, ge=0, description="Number of tools in the resolved set"
)
capability_summary: SkillCapabilitySummary = Field(
default_factory=lambda: SkillCapabilitySummary.model_validate({}),
description="Aggregated capability summary",
)
source_types: list[str] = Field(
default_factory=list,
description="Distinct source type labels present in the skill",
)
writes: bool = Field(
default=False,
description="Whether any tool in the skill can write",
)
read_only: bool = Field(
default=True,
description="Whether the skill is entirely read-only",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
validate_assignment=True,
)
# -- Factory ---------------------------------------------------------------
@classmethod
def from_skill(
cls,
skill: Skill,
resolved: list[ResolvedToolEntry] | None = None,
version: str = "0.0.0",
) -> SkillMetadata:
"""Create metadata from a ``Skill`` domain model.
If *resolved* entries are not provided, the skill is resolved
using a fresh ``SkillResolver`` with an empty lookup.
Args:
skill: The domain-level ``Skill`` instance.
resolved: Pre-resolved tool entries (optional).
version: Semantic version string for the skill.
Returns:
A new ``SkillMetadata`` instance.
"""
if skill is None:
raise ValueError("skill must not be None")
resolver = SkillResolver()
if resolved is None:
resolved = resolver.resolve_tools(skill, {})
summary = resolver.compute_capability_summary(skill, resolved)
# Derive source_types from resolved entries
source_types: list[str] = []
for entry in resolved:
if entry.is_inline:
label = "inline"
elif entry.name.startswith("mcp:"):
label = "mcp"
elif entry.name.startswith("agent_skill:"):
label = "agent_skill"
else:
label = "tool_ref"
if label not in source_types:
source_types.append(label)
has_writes = summary.write_tools > 0 or summary.has_side_effects
is_read_only = summary.write_tools == 0 and not summary.has_side_effects
return cls(
name=skill.name,
description=skill.description,
version=version,
tool_count=summary.total_tools,
capability_summary=summary,
source_types=source_types,
writes=has_writes,
read_only=is_read_only,
)
# ---------------------------------------------------------------------------
# SkillDefinition
# ---------------------------------------------------------------------------
class SkillDefinition(BaseModel):
"""Full skill definition with resolved tools and metadata.
Pairs the domain ``Skill`` model with the resolved (flattened) tool
entries and pre-computed ``SkillMetadata``. Includes validation
that ensures schema consistency between the definition's
input/output schemas and the Tool Registry schemas.
"""
skill: Skill = Field(..., description="Domain-level Skill model")
resolved_tools: list[ResolvedToolEntry] = Field(
default_factory=list,
description="Flattened set of resolved tool entries",
)
metadata: SkillMetadata = Field(..., description="Pre-computed skill metadata")
input_schema: dict[str, Any] | None = Field(
default=None,
description="JSON Schema for skill-level inputs",
)
output_schema: dict[str, Any] | None = Field(
default=None,
description="JSON Schema for skill-level outputs",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
validate_assignment=True,
)
# -- Schema validation -----------------------------------------------------
@model_validator(mode="after")
def _validate_schemas(self) -> SkillDefinition:
"""Validate that input/output schemas are well-formed JSON Schema.
A valid JSON Schema object must have ``"type"`` declared.
This lightweight check catches the most common authoring errors
without pulling in a full JSON Schema meta-validator.
"""
for label, schema in [
("input_schema", self.input_schema),
("output_schema", self.output_schema),
]:
if schema is not None:
if not isinstance(schema, dict):
raise ValueError(f"{label} must be a dict, got {type(schema)}")
if "type" not in schema:
raise ValueError(
f"{label} must include a 'type' key to be a valid JSON Schema"
)
return self
# -- Writes / read_only propagation ----------------------------------------
@model_validator(mode="after")
def _validate_writes_consistency(self) -> SkillDefinition:
"""Ensure writes/read_only metadata is consistent.
If metadata declares ``read_only=True`` but the resolved tools
include write-capable inline tools, the skill is inconsistent
and we raise a validation error.
"""
write_inline_count = 0
for entry in self.resolved_tools:
if (
entry.is_inline
and entry.inline_tool is not None
and entry.inline_tool.capability is not None
and entry.inline_tool.capability.writes
):
write_inline_count += 1
if self.metadata.read_only and write_inline_count > 0:
raise ValueError(
"SkillDefinition metadata declares read_only=True but "
f"{write_inline_count} resolved inline tool(s) have "
"writes=True"
)
return self
# ---------------------------------------------------------------------------
# SkillResult
# ---------------------------------------------------------------------------
class SkillResult(BaseModel):
"""Result of a single skill-tool invocation.
Captures success/failure, output data, timing, and any resource
changes produced by the tool. An optional ``SkillError`` provides
structured diagnostics on failure.
"""
skill_name: str = Field(..., min_length=1, description="Skill that was invoked")
tool_name: str = Field(..., min_length=1, description="Tool that was invoked")
success: bool = Field(..., description="Whether the invocation succeeded")
output_data: Any = Field(default=None, description="Output data from the tool")
error: SkillError | None = Field(
default=None,
description="Structured error on failure",
)
duration_ms: float = Field(
default=0.0, ge=0.0, description="Execution duration in milliseconds"
)
resource_changes: list[dict[str, Any]] = Field(
default_factory=list,
description="List of resource change descriptors",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# Error mapping helpers
# ---------------------------------------------------------------------------
def map_tool_error(
exc: Exception,
skill_name: str,
tool_name: str | None = None,
) -> SkillError:
"""Normalise an arbitrary exception into a ``SkillError``.
The mapping heuristic inspects the exception message and type to
select the most appropriate ``SkillErrorType``:
- ``ValueError`` with *"cycle"* → ``CYCLE_DETECTED``
- ``ValueError`` with *"not found"* → ``SKILL_NOT_FOUND``
- ``PermissionError`` → ``PERMISSION_DENIED``
- ``ValidationError`` (Pydantic) → ``VALIDATION_ERROR``
- Everything else → ``TOOL_EXECUTION_FAILURE``
Args:
exc: The exception to map.
skill_name: Name of the skill context.
tool_name: Optional tool name context.
Returns:
A new ``SkillError`` instance.
Raises:
ValueError: If *skill_name* is empty.
"""
if not skill_name:
raise ValueError("skill_name must not be empty")
msg = str(exc)
error_type = _classify_exception(exc, msg)
details: dict[str, Any] = {
"exception_type": type(exc).__name__,
}
if tool_name:
details["tool_name"] = tool_name
return SkillError(
error_type=error_type,
message=msg or f"Unknown error: {type(exc).__name__}",
details=details,
skill_name=skill_name,
tool_name=tool_name,
)
def _classify_exception(exc: Exception, msg: str) -> SkillErrorType:
"""Classify an exception into a ``SkillErrorType``.
Args:
exc: The exception instance.
msg: String representation of the exception.
Returns:
The determined ``SkillErrorType``.
"""
lower_msg = msg.lower()
if isinstance(exc, ValueError):
if "cycle" in lower_msg:
return SkillErrorType.CYCLE_DETECTED
if "not found" in lower_msg:
return SkillErrorType.SKILL_NOT_FOUND
return SkillErrorType.RESOLUTION_FAILURE
if isinstance(exc, PermissionError):
return SkillErrorType.PERMISSION_DENIED
# Pydantic ValidationError check (avoid importing pydantic here to
# keep the mapping function lightweight)
if type(exc).__name__ == "ValidationError":
return SkillErrorType.VALIDATION_ERROR
if "activation" in lower_msg or "activate" in lower_msg:
return SkillErrorType.TOOL_ACTIVATION_FAILURE
return SkillErrorType.TOOL_EXECUTION_FAILURE