Compare commits

...

3 Commits

Author SHA1 Message Date
freemo 9361025659 feat(skill): add inline tool executor
CI / lint (pull_request) Successful in 15s
CI / build (pull_request) Successful in 17s
CI / quality (pull_request) Successful in 22s
CI / typecheck (pull_request) Successful in 40s
CI / security (pull_request) Successful in 42s
CI / integration_tests (pull_request) Failing after 2m17s
CI / unit_tests (pull_request) Successful in 6m4s
CI / docker (pull_request) Successful in 38s
CI / coverage (pull_request) Successful in 12m10s
2026-02-19 09:03:59 +00:00
freemo 94a78b3311 feat(skill): add skill context and registry
CI / lint (pull_request) Successful in 16s
CI / build (pull_request) Successful in 18s
CI / quality (pull_request) Successful in 19s
CI / typecheck (pull_request) Successful in 29s
CI / security (pull_request) Successful in 29s
CI / integration_tests (pull_request) Failing after 2m8s
CI / unit_tests (pull_request) Successful in 3m39s
CI / docker (pull_request) Successful in 39s
CI / coverage (pull_request) Successful in 9m44s
2026-02-19 02:15:56 +00:00
freemo 6e440ba4db feat(skill): add skill protocol and metadata
CI / lint (pull_request) Successful in 20s
CI / build (pull_request) Successful in 19s
CI / quality (pull_request) Successful in 33s
CI / typecheck (pull_request) Successful in 40s
CI / security (pull_request) Successful in 54s
CI / integration_tests (pull_request) Failing after 2m18s
CI / unit_tests (pull_request) Successful in 5m37s
CI / docker (pull_request) Successful in 1m1s
CI / coverage (pull_request) Successful in 25m1s
2026-02-19 00:38:10 +00:00
25 changed files with 4808 additions and 59 deletions
+118
View File
@@ -0,0 +1,118 @@
"""ASV benchmarks for inline tool executor operations.
Measures the performance of:
- InlineToolExecutor creation
- Simple inline tool execution overhead
- InlineToolResult construction
- Tool validation
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.domain.models.core.skill import SkillInlineTool
from cleveragents.domain.models.core.tool import ToolCapability, ToolSource
from cleveragents.skills.context import SkillContext
from cleveragents.skills.inline_executor import InlineToolExecutor, InlineToolResult
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.skill import SkillInlineTool
from cleveragents.domain.models.core.tool import ToolCapability, ToolSource
from cleveragents.skills.context import SkillContext
from cleveragents.skills.inline_executor import InlineToolExecutor, InlineToolResult
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
_SANDBOX = Path("/tmp/bench-inline-sandbox")
_SIMPLE_TOOL = SkillInlineTool(
description="Benchmark print tool",
source=ToolSource.CUSTOM,
code='print("bench")',
)
_NOOP_TOOL = SkillInlineTool(
description="Benchmark no-op tool",
source=ToolSource.CUSTOM,
code="pass",
)
_WRITE_TOOL = SkillInlineTool(
description="Benchmark write tool",
source=ToolSource.CUSTOM,
code='print("write")',
capability=ToolCapability(writes=True),
)
# ---------------------------------------------------------------------------
# Executor Benchmarks
# ---------------------------------------------------------------------------
class TimeInlineExecutorCreation:
"""Benchmark InlineToolExecutor creation throughput."""
timeout = 60
def time_create_executor(self) -> None:
"""Time basic InlineToolExecutor creation."""
InlineToolExecutor()
def time_create_executor_custom(self) -> None:
"""Time InlineToolExecutor creation with custom limits."""
InlineToolExecutor(max_runtime_seconds=10.0, max_output_bytes=512)
class TimeInlineToolExecution:
"""Benchmark inline tool execution overhead."""
timeout = 60
def setup(self) -> None:
"""Create executor and context for benchmarking."""
self.executor = InlineToolExecutor()
self.ctx = SkillContext(
plan_id="plan-bench-inline",
project_id="proj-bench-inline",
sandbox_path=_SANDBOX,
)
def time_execute_noop(self) -> None:
"""Time no-op inline tool execution."""
self.executor.execute(_NOOP_TOOL, self.ctx, {})
def time_execute_print(self) -> None:
"""Time simple print inline tool execution."""
self.executor.execute(_SIMPLE_TOOL, self.ctx, {})
def time_validate_tool(self) -> None:
"""Time tool validation."""
self.executor.validate_tool(_SIMPLE_TOOL)
class TimeInlineToolResult:
"""Benchmark InlineToolResult construction."""
timeout = 60
def time_create_success_result(self) -> None:
"""Time successful result creation."""
InlineToolResult(
success=True,
output="benchmark output",
duration_ms=1.5,
)
def time_create_error_result(self) -> None:
"""Time error result creation."""
InlineToolResult(
success=False,
error_message="benchmark error",
duration_ms=2.0,
)
+174
View File
@@ -0,0 +1,174 @@
"""ASV benchmarks for Skill context and registry operations.
Measures the performance of:
- SkillContext creation and metadata access
- SkillContext resource resolution
- SkillContext change tracking
- SkillContext write guard enforcement
- SkillRegistry register/get/list/unregister
- SkillRegistry resolve_tools
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.domain.models.core.skill import (
Skill,
SkillResolver,
)
from cleveragents.skills.context import SkillContext
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillMetadata,
)
from cleveragents.skills.registry import SkillRegistry
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.skill import (
Skill,
SkillResolver,
)
from cleveragents.skills.context import SkillContext
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillMetadata,
)
from cleveragents.skills.registry import SkillRegistry
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
_SANDBOX = Path("/tmp/bench-sandbox")
_BASIC_SKILL = Skill(
name="local/bench-ctx-skill",
description="Benchmark context skill",
tool_refs=["local/tool-a", "local/tool-b"],
)
_RESOLVER = SkillResolver()
_RESOLVED = _RESOLVER.resolve_tools(_BASIC_SKILL, {})
_METADATA = SkillMetadata.from_skill(_BASIC_SKILL, resolved=_RESOLVED)
_DEFINITION = SkillDefinition(
skill=_BASIC_SKILL,
resolved_tools=_RESOLVED,
metadata=_METADATA,
)
# ---------------------------------------------------------------------------
# Context Benchmarks
# ---------------------------------------------------------------------------
class TimeSkillContextCreation:
"""Benchmark SkillContext creation throughput."""
timeout = 60
def time_create_context(self) -> None:
"""Time basic SkillContext creation."""
SkillContext(
plan_id="plan-bench",
project_id="proj-bench",
sandbox_path=_SANDBOX,
)
def time_create_context_with_bindings(self) -> None:
"""Time SkillContext creation with resource bindings."""
SkillContext(
plan_id="plan-bench",
project_id="proj-bench",
sandbox_path=_SANDBOX,
resource_bindings={
"git-checkout": "/repo/main",
"artifact-store": "/artifacts",
},
)
class TimeSkillContextOperations:
"""Benchmark SkillContext operations."""
timeout = 60
def setup(self) -> None:
"""Create a context for benchmarking."""
self.ctx = SkillContext(
plan_id="plan-ops",
project_id="proj-ops",
sandbox_path=_SANDBOX,
resource_bindings={"git-checkout": "/repo"},
)
def time_resolve_resource(self) -> None:
"""Time resource resolution."""
self.ctx.resolve_resource("git-checkout")
def time_get_plan_metadata(self) -> None:
"""Time plan metadata retrieval."""
self.ctx.get_plan_metadata()
def time_register_invocation(self) -> None:
"""Time tool invocation registration."""
self.ctx.register_tool_invocation("local/bench-tool", {"a": 1}, {"b": 2}, 5.0)
def time_enforce_write_guard_pass(self) -> None:
"""Time write guard check (non-read-only context)."""
self.ctx.enforce_write_guard("local/bench-tool")
# ---------------------------------------------------------------------------
# Registry Benchmarks
# ---------------------------------------------------------------------------
class TimeSkillRegistryOperations:
"""Benchmark SkillRegistry operations."""
timeout = 60
def setup(self) -> None:
"""Create a populated registry for benchmarking."""
self.registry = SkillRegistry()
for i in range(50):
skill = Skill(
name=f"local/bench-skill-{i:03d}",
description=f"Benchmark skill {i}",
tool_refs=["local/tool-a", "local/tool-b"],
)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
meta = SkillMetadata.from_skill(skill, resolved=resolved)
defn = SkillDefinition(skill=skill, resolved_tools=resolved, metadata=meta)
self.registry.register(defn)
def time_get_skill(self) -> None:
"""Time skill lookup."""
self.registry.get("local/bench-skill-025")
def time_list_all(self) -> None:
"""Time listing all skills."""
self.registry.list_all()
def time_resolve_tools(self) -> None:
"""Time tool resolution for a skill."""
self.registry.resolve_tools("local/bench-skill-010")
def time_register_and_unregister(self) -> None:
"""Time register + unregister cycle."""
name = "local/bench-temp"
skill = Skill(name=name, description="Temp skill")
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
meta = SkillMetadata.from_skill(skill, resolved=resolved)
defn = SkillDefinition(skill=skill, resolved_tools=resolved, metadata=meta)
self.registry.register(defn)
self.registry.unregister(name)
+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",
)
+97
View File
@@ -0,0 +1,97 @@
# Skill Context & Registry
Runtime context and registry for skill execution in CleverAgents v3.
## SkillContext
The `SkillContext` class provides the runtime environment for skill
execution. It carries plan/project identifiers, a sandbox root path,
resource bindings, and a change tracker for recording tool invocations.
### Fields
| Field | Type | Description |
|-------|------|-------------|
| `plan_id` | `str` | Identifier for the current plan |
| `project_id` | `str` | Identifier for the current project |
| `sandbox_path` | `Path` | Root path to the execution sandbox |
| `change_tracker` | `list[dict]` | Recorded tool invocation records |
| `resource_bindings` | `dict[str, Any]` | Bound resource name → value mapping |
| `read_only` | `bool` | Whether write operations are forbidden |
| `metadata` | `dict[str, Any]` | Arbitrary plan/project metadata |
### Methods
#### `resolve_resource(name: str) -> Any`
Look up a bound resource by name from `resource_bindings`. Raises
`SkillError(RESOLUTION_FAILURE)` if the resource is not found.
#### `get_plan_metadata() -> dict[str, Any]`
Return a dict containing `plan_id`, `project_id`, and any additional
metadata stored on the context.
#### `get_sandbox_path() -> Path`
Return the sandbox root `Path`.
#### `is_read_only() -> bool`
Check whether this context forbids write operations.
#### `register_tool_invocation(tool_name, input_data, output_data, duration_ms)`
Record a tool invocation in the change tracker. Each record stores
the tool name, input/output data, duration, plan ID, and project ID.
#### `enforce_write_guard(tool_name: str)`
Raise `SkillError(PERMISSION_DENIED)` if the context is read-only and
a tool attempts to write. Should be called before any write operation.
## SkillRegistry
The `SkillRegistry` class manages in-memory registration and lookup of
skill definitions.
### Constructor
```python
SkillRegistry(tool_registry=None)
```
Optionally accepts a reference to a Tool Registry service for
tool-ref validation.
### Methods
#### `register(skill: SkillDefinition)`
Register a skill definition. Raises `SkillError(VALIDATION_ERROR)` if
a skill with the same name is already registered.
#### `unregister(name: str)`
Remove a skill from the registry. Raises `SkillError(SKILL_NOT_FOUND)`
if the skill is not registered.
#### `get(name: str) -> SkillDefinition`
Look up a skill by name. Raises `SkillError(SKILL_NOT_FOUND)` if not
found.
#### `list_all() -> list[SkillMetadata]`
Return metadata for all registered skills, sorted by name.
#### `resolve_tools(skill_name: str) -> list[ResolvedToolEntry]`
Resolve tool references for a registered skill using the
`SkillResolver`.
#### `validate_skill(skill: SkillDefinition) -> list[str]`
Validate tool references exist in the tool registry, inline tools have
descriptions, and included skills are registered. Returns a list of
error messages (empty means valid).
+111
View File
@@ -0,0 +1,111 @@
# Inline Tool Executor
Safe execution of inline (anonymous) tool code within skills.
## Overview
The `InlineToolExecutor` runs Python code strings defined directly in
skill YAML. Execution is bounded by configurable timeout and output-size
limits, with write-guard enforcement delegated to the `SkillContext`.
## InlineToolResult
Frozen Pydantic model returned from every execution.
| Field | Type | Description |
|-------|------|-------------|
| `success` | `bool` | Whether the execution succeeded |
| `output` | `Any` | Captured stdout/stderr output |
| `error_message` | `str \| None` | Error description on failure |
| `duration_ms` | `float` | Wall-clock execution time in milliseconds |
| `truncated` | `bool` | Whether output was truncated due to size limits |
## InlineToolExecutor
### Constructor
```python
InlineToolExecutor(
max_runtime_seconds: float = 30.0,
max_output_bytes: int = 1_048_576,
)
```
| Parameter | Default | Description |
|-----------|---------|-------------|
| `max_runtime_seconds` | `30.0` | Maximum wall-clock seconds for execution |
| `max_output_bytes` | `1_048_576` | Maximum captured output before truncation (1 MiB) |
### `execute(tool, context, input_data) -> InlineToolResult`
Run an inline tool's code with safety guards.
1. **Validation** -- Checks the tool has code and uses a supported source.
2. **Write guard** -- If `tool.capability.writes` is `True`, calls
`context.enforce_write_guard()`. In a read-only context this
returns an error result without executing.
3. **Path restriction** -- File-path keys in `input_data` (keys ending
with `_path`, `_file`, or equal to `path`) are validated against
`context.get_sandbox_path()`.
4. **Threaded execution** -- Code runs in a daemon thread with
`max_runtime_seconds` timeout.
5. **Output capture** -- stdout and stderr are captured. If the
combined output exceeds `max_output_bytes`, it is truncated and
`truncated=True` is set on the result.
6. **Change tracking** -- The invocation is recorded in the context's
change tracker regardless of success/failure.
### `validate_tool(tool) -> list[str]`
Return a list of validation errors (empty means valid).
Checks:
- Tool has non-empty `code`.
- Tool's `source` is `custom` (Python). Other source types are not
supported for MVP.
## Safety Constraints
### Timeout
Execution uses `threading.Thread` with a join timeout. If the thread
does not complete within `max_runtime_seconds`, the result reports:
```
Execution timed out after {max_runtime_seconds}s
```
The daemon thread is abandoned (not forcibly killed). For MVP this is
acceptable; future versions may use process isolation.
### Output Size
Combined stdout + stderr is capped at `max_output_bytes`. When
exceeded the output is byte-truncated and `InlineToolResult.truncated`
is set to `True`.
### Write Guard
Tools declaring `capability.writes = True` trigger a write-guard check
via `SkillContext.enforce_write_guard()`. In a read-only context the
executor returns an error result **without executing any code**.
### Path Restriction
Input data keys matching `*_path`, `*_file`, or `path` are resolved
and checked against the sandbox root. Paths that escape the sandbox
are rejected before execution.
### Language / Source Support
Only `source = custom` (Python) is supported for MVP. Other source
types (`mcp`, `agent_skill`, `builtin`, `wrapped`) are rejected by
`validate_tool`.
### Network Access
For MVP, inline tools run in the host process and **no network
sandboxing** is enforced. Tools should be treated as local-only.
Future milestones will introduce process-level or container-level
isolation for network restrictions.
+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).
+140
View File
@@ -0,0 +1,140 @@
Feature: Skill Context and Registry
As a developer
I want a runtime context for skill execution and a skill registry
So that skills can access plan/project info, track changes, and be managed centrally
# ---- SkillContext Creation ----
Scenario: Create SkillContext with plan/project/sandbox info
When I create a skill_context with plan "plan-001" project "proj-alpha" and sandbox "/tmp/sandbox"
Then the skill_context should be created
And the skill_context plan_id should be "plan-001"
And the skill_context project_id should be "proj-alpha"
And the skill_context sandbox_path should be "/tmp/sandbox"
And the skill_context should not be read_only
Scenario: Create read-only SkillContext
When I create a read_only skill_context with plan "plan-002" project "proj-beta"
Then the skill_context should be created
And the skill_context should be read_only
Scenario: SkillContext with custom metadata
When I create a skill_context with metadata key "env" value "staging"
Then the skill_context plan metadata should contain key "env"
And the skill_context plan metadata key "env" should be "staging"
# ---- Resource Resolution ----
Scenario: Resolve bound resources from context
When I create a skill_context with resource "git-checkout" bound to "/repo/checkout"
And I resolve resource "git-checkout" from the skill_context
Then the resolved resource should be "/repo/checkout"
Scenario: Resolve missing resource raises SkillError
When I create a skill_context with no resources
And I try to resolve resource "missing-resource" from the skill_context
Then a skill_context resolution error should be raised
# ---- Change Tracking ----
Scenario: Register tool invocation in change tracker
When I create a skill_context with plan "plan-003" project "proj-gamma" and sandbox "/tmp/sb"
And I register a tool invocation for "local/edit-file" with duration 42.5
Then the skill_context change tracker should have 1 records
And the last change tracker record tool_name should be "local/edit-file"
And the last change tracker record duration_ms should be 42.5
Scenario: Multiple tool invocations are tracked in order
When I create a skill_context with plan "plan-004" project "proj-delta" and sandbox "/tmp/sb2"
And I register a tool invocation for "local/read-file" with duration 10.0
And I register a tool invocation for "local/write-file" with duration 20.0
Then the skill_context change tracker should have 2 records
# ---- Write Guard ----
Scenario: Enforce read-only write guard raises SkillError
When I create a read_only skill_context with plan "plan-005" project "proj-epsilon"
And I try to enforce write guard for tool "local/write-file"
Then a skill_context permission denied error should be raised
And the skill_context permission error tool_name should be "local/write-file"
Scenario: Write guard passes in writable context
When I create a skill_context with plan "plan-006" project "proj-zeta" and sandbox "/tmp/sb3"
And I enforce write guard for tool "local/write-file" in writable context
Then no skill_context error should be raised
# ---- Plan Metadata ----
Scenario: Get plan metadata includes plan_id and project_id
When I create a skill_context with plan "plan-007" project "proj-eta" and sandbox "/tmp/sb4"
Then the skill_context plan metadata should contain key "plan_id"
And the skill_context plan metadata key "plan_id" should be "plan-007"
And the skill_context plan metadata should contain key "project_id"
And the skill_context plan metadata key "project_id" should be "proj-eta"
# ---- SkillRegistry Register/Get/List/Unregister ----
Scenario: SkillRegistry register and get skill
When I create a skill_registry
And I register a skill "local/test-reg" in the registry
And I get skill "local/test-reg" from the registry
Then the registry skill name should be "local/test-reg"
Scenario: SkillRegistry register duplicate raises error
When I create a skill_registry
And I register a skill "local/dup-skill" in the registry
And I try to register a duplicate skill "local/dup-skill"
Then a skill_context validation error should be raised
Scenario: SkillRegistry get missing skill raises error
When I create a skill_registry
And I try to get skill "local/missing" from the registry
Then a skill_context not found error should be raised
Scenario: SkillRegistry list_all returns metadata
When I create a skill_registry
And I register a skill "local/skill-a" in the registry
And I register a skill "local/skill-b" in the registry
And I list all skills from the registry
Then the registry should have 2 skills
And the registry skill list should contain "local/skill-a"
And the registry skill list should contain "local/skill-b"
Scenario: SkillRegistry unregister removes skill
When I create a skill_registry
And I register a skill "local/to-remove" in the registry
And I unregister skill "local/to-remove" from the registry
And I try to get skill "local/to-remove" from the registry
Then a skill_context not found error should be raised
Scenario: SkillRegistry unregister missing skill raises error
When I create a skill_registry
And I try to unregister skill "local/nonexistent" from the registry
Then a skill_context not found error should be raised
# ---- SkillRegistry Tool Resolution ----
Scenario: SkillRegistry resolve_tools through tool references
When I create a skill_registry
And I register a skill "local/resolve-test" with tool refs in the registry
And I resolve tools for "local/resolve-test" from the registry
Then the resolved tools should have 2 entries
And the resolved tools should contain "local/tool-a"
And the resolved tools should contain "local/tool-b"
# ---- SkillRegistry Validation ----
Scenario: SkillRegistry validate_skill detects missing tools
When I create a skill_registry with a mock tool registry
And I validate a skill with missing tool refs
Then the validation errors should contain "not found in tool registry"
Scenario: SkillRegistry validate_skill detects missing includes
When I create a skill_registry
And I validate a skill with missing includes
Then the validation errors should contain "is not registered"
Scenario: SkillRegistry validate_skill returns empty for valid skill
When I create a skill_registry
And I validate a skill with no issues
Then the validation errors should be empty
+85
View File
@@ -0,0 +1,85 @@
Feature: Inline Tool Executor
As a developer
I want to execute inline tool code with safety constraints
So that skills can run embedded Python code safely and reliably
# ---- Successful Execution ----
Scenario: Execute simple inline tool successfully
Given an inline tool with code that prints "hello inline"
And a writable skill context with sandbox "/tmp/inline-sandbox"
When I execute the inline tool
Then the inline result should be successful
And the inline result output should contain "hello inline"
Scenario: Inline tool returns output data
Given an inline tool with code that prints "result: 42"
And a writable skill context with sandbox "/tmp/inline-sandbox"
When I execute the inline tool
Then the inline result should be successful
And the inline result output should contain "result: 42"
# ---- Timeout ----
Scenario: Inline tool times out and returns structured error
Given an inline tool with code that sleeps for 5 seconds
And a writable skill context with sandbox "/tmp/inline-sandbox"
And an inline executor with max runtime 0.1 seconds
When I execute the inline tool
Then the inline result should not be successful
And the inline result error message should contain "timed out"
# ---- Output Truncation ----
Scenario: Inline tool output exceeds max size and is truncated
Given an inline tool with code that prints 2000 bytes of output
And a writable skill context with sandbox "/tmp/inline-sandbox"
And an inline executor with max output 100 bytes
When I execute the inline tool
Then the inline result should be successful
And the inline result should be truncated
# ---- Write Guard ----
Scenario: Read-only context blocks inline tool with writes capability
Given an inline tool with writes capability and code that prints "write op"
And a read-only skill context with sandbox "/tmp/inline-sandbox"
When I execute the inline tool
Then the inline result should not be successful
And the inline result error message should contain "denied"
Scenario: Writable context allows inline tool with writes capability
Given an inline tool with writes capability and code that prints "write op"
And a writable skill context with sandbox "/tmp/inline-sandbox"
When I execute the inline tool
Then the inline result should be successful
# ---- Validation ----
Scenario: Validate inline tool with missing code
Given an inline tool with no code
When I validate the inline tool
Then the inline validation errors should contain "non-empty code"
Scenario: Validate inline tool with unsupported language
Given an inline tool with unsupported source type
When I validate the inline tool
Then the inline validation errors should contain "Unsupported source"
# ---- Change Tracking ----
Scenario: Tool invocation is recorded in change tracker
Given an inline tool with code that prints "tracked"
And a writable skill context with sandbox "/tmp/inline-sandbox"
When I execute the inline tool
Then the skill context change tracker should have 1 inline records
And the last inline tracker record tool_name should be "inline_tool"
# ---- Error Handling ----
Scenario: Inline tool execution error returns structured result
Given an inline tool with code that raises an exception
And a writable skill context with sandbox "/tmp/inline-sandbox"
When I execute the inline tool
Then the inline result should not be successful
And the inline result error message should contain "intentional error"
+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
+484
View File
@@ -0,0 +1,484 @@
"""Step definitions for Skill context and registry tests."""
from __future__ import annotations
from pathlib import Path
from typing import Any
from unittest.mock import MagicMock
from behave import then, when
from behave.runner import Context
from cleveragents.domain.models.core.skill import (
Skill,
SkillResolver,
)
from cleveragents.skills.context import SkillContext, SkillExecutionError
from cleveragents.skills.protocol import (
SkillDefinition,
SkillMetadata,
)
from cleveragents.skills.registry import SkillRegistry
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_skill_definition(name: str, **overrides: Any) -> SkillDefinition:
"""Create a SkillDefinition with sensible defaults."""
defaults: dict[str, Any] = {
"name": name,
"description": f"Test skill {name}",
}
defaults.update(overrides)
skill = Skill(**defaults)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
metadata = SkillMetadata.from_skill(skill, resolved=resolved)
return SkillDefinition(
skill=skill,
resolved_tools=resolved,
metadata=metadata,
)
# ---------------------------------------------------------------------------
# SkillContext Creation
# ---------------------------------------------------------------------------
@when(
'I create a skill_context with plan "{plan_id}" project "{project_id}" and sandbox "{sandbox}"'
)
def skill_context_create(
context: Context, plan_id: str, project_id: str, sandbox: str
) -> None:
"""Create a SkillContext with plan/project/sandbox info."""
context.skill_context = SkillContext(
plan_id=plan_id,
project_id=project_id,
sandbox_path=Path(sandbox),
)
context.skill_context_error = None
@when('I create a read_only skill_context with plan "{plan_id}" project "{project_id}"')
def skill_context_create_readonly(
context: Context, plan_id: str, project_id: str
) -> None:
"""Create a read-only SkillContext."""
context.skill_context = SkillContext(
plan_id=plan_id,
project_id=project_id,
sandbox_path=Path("/tmp/readonly-sandbox"),
read_only=True,
)
context.skill_context_error = None
@when('I create a skill_context with metadata key "{key}" value "{value}"')
def skill_context_create_with_metadata(context: Context, key: str, value: str) -> None:
"""Create a SkillContext with custom metadata."""
context.skill_context = SkillContext(
plan_id="plan-meta",
project_id="proj-meta",
sandbox_path=Path("/tmp/meta-sandbox"),
metadata={key: value},
)
@then("the skill_context should be created")
def skill_context_check_created(context: Context) -> None:
"""Verify context was created."""
assert context.skill_context is not None
@then('the skill_context plan_id should be "{expected}"')
def skill_context_check_plan_id(context: Context, expected: str) -> None:
"""Check context plan_id."""
assert context.skill_context.plan_id == expected
@then('the skill_context project_id should be "{expected}"')
def skill_context_check_project_id(context: Context, expected: str) -> None:
"""Check context project_id."""
assert context.skill_context.project_id == expected
@then('the skill_context sandbox_path should be "{expected}"')
def skill_context_check_sandbox(context: Context, expected: str) -> None:
"""Check context sandbox_path."""
assert str(context.skill_context.get_sandbox_path()) == expected
@then("the skill_context should not be read_only")
def skill_context_check_not_readonly(context: Context) -> None:
"""Check context is not read-only."""
assert context.skill_context.is_read_only() is False
@then("the skill_context should be read_only")
def skill_context_check_readonly(context: Context) -> None:
"""Check context is read-only."""
assert context.skill_context.is_read_only() is True
# ---------------------------------------------------------------------------
# Plan Metadata
# ---------------------------------------------------------------------------
@then('the skill_context plan metadata should contain key "{key}"')
def skill_context_check_metadata_key(context: Context, key: str) -> None:
"""Check plan metadata contains a key."""
meta = context.skill_context.get_plan_metadata()
assert key in meta, f"Expected '{key}' in {list(meta.keys())}"
@then('the skill_context plan metadata key "{key}" should be "{expected}"')
def skill_context_check_metadata_value(
context: Context, key: str, expected: str
) -> None:
"""Check plan metadata value."""
meta = context.skill_context.get_plan_metadata()
actual = meta[key]
assert str(actual) == expected, f"Expected '{expected}', got '{actual}'"
# ---------------------------------------------------------------------------
# Resource Resolution
# ---------------------------------------------------------------------------
@when('I create a skill_context with resource "{name}" bound to "{value}"')
def skill_context_create_with_resource(context: Context, name: str, value: str) -> None:
"""Create a SkillContext with a resource binding."""
context.skill_context = SkillContext(
plan_id="plan-res",
project_id="proj-res",
sandbox_path=Path("/tmp/res-sandbox"),
resource_bindings={name: value},
)
@when('I resolve resource "{name}" from the skill_context')
def skill_context_resolve_resource(context: Context, name: str) -> None:
"""Resolve a resource from the context."""
context.resolved_resource = context.skill_context.resolve_resource(name)
@then('the resolved resource should be "{expected}"')
def skill_context_check_resolved_resource(context: Context, expected: str) -> None:
"""Check resolved resource value."""
assert context.resolved_resource == expected
@when("I create a skill_context with no resources")
def skill_context_create_no_resources(context: Context) -> None:
"""Create a SkillContext with empty resource bindings."""
context.skill_context = SkillContext(
plan_id="plan-empty",
project_id="proj-empty",
sandbox_path=Path("/tmp/empty-sandbox"),
)
@when('I try to resolve resource "{name}" from the skill_context')
def skill_context_try_resolve_resource(context: Context, name: str) -> None:
"""Try to resolve a missing resource."""
context.skill_context_error = None
try:
context.skill_context.resolve_resource(name)
except SkillExecutionError as e:
context.skill_context_error = e
@then("a skill_context resolution error should be raised")
def skill_context_check_resolution_error(context: Context) -> None:
"""Verify resolution error was raised."""
assert context.skill_context_error is not None
assert context.skill_context_error.error_type.value == "resolution_failure"
# ---------------------------------------------------------------------------
# Change Tracking
# ---------------------------------------------------------------------------
@when('I register a tool invocation for "{tool_name}" with duration {duration:g}')
def skill_context_register_invocation(
context: Context, tool_name: str, duration: float
) -> None:
"""Register a tool invocation."""
context.skill_context.register_tool_invocation(
tool_name=tool_name,
input_data={"arg": "test"},
output_data={"result": "ok"},
duration_ms=duration,
)
@then("the skill_context change tracker should have {count:d} records")
def skill_context_check_tracker_count(context: Context, count: int) -> None:
"""Check change tracker record count."""
actual = len(context.skill_context.change_tracker)
assert actual == count, f"Expected {count} records, got {actual}"
@then('the last change tracker record tool_name should be "{expected}"')
def skill_context_check_last_tool(context: Context, expected: str) -> None:
"""Check last tracker record tool_name."""
record = context.skill_context.change_tracker[-1]
assert record["tool_name"] == expected
@then("the last change tracker record duration_ms should be {expected:g}")
def skill_context_check_last_duration(context: Context, expected: float) -> None:
"""Check last tracker record duration."""
record = context.skill_context.change_tracker[-1]
assert record["duration_ms"] == expected
# ---------------------------------------------------------------------------
# Write Guard
# ---------------------------------------------------------------------------
@when('I try to enforce write guard for tool "{tool_name}"')
def skill_context_try_write_guard(context: Context, tool_name: str) -> None:
"""Try to enforce write guard on a read-only context."""
context.skill_context_error = None
try:
context.skill_context.enforce_write_guard(tool_name)
except SkillExecutionError as e:
context.skill_context_error = e
@then("a skill_context permission denied error should be raised")
def skill_context_check_permission_error(context: Context) -> None:
"""Verify permission denied error was raised."""
assert context.skill_context_error is not None
assert context.skill_context_error.error_type.value == "permission_denied"
@then('the skill_context permission error tool_name should be "{expected}"')
def skill_context_check_permission_tool(context: Context, expected: str) -> None:
"""Check permission error tool_name."""
assert context.skill_context_error.tool_name == expected
@when('I enforce write guard for tool "{tool_name}" in writable context')
def skill_context_write_guard_writable(context: Context, tool_name: str) -> None:
"""Enforce write guard in a writable context (should not raise)."""
context.skill_context_error = None
try:
context.skill_context.enforce_write_guard(tool_name)
except SkillExecutionError as e:
context.skill_context_error = e
@then("no skill_context error should be raised")
def skill_context_check_no_error(context: Context) -> None:
"""Verify no error was raised."""
assert context.skill_context_error is None
# ---------------------------------------------------------------------------
# SkillRegistry
# ---------------------------------------------------------------------------
@when("I create a skill_registry")
def skill_registry_create(context: Context) -> None:
"""Create a SkillRegistry."""
context.skill_registry = SkillRegistry()
context.skill_context_error = None
@when('I register a skill "{name}" in the registry')
def skill_registry_register(context: Context, name: str) -> None:
"""Register a skill in the registry."""
defn = _make_skill_definition(name)
context.skill_registry.register(defn)
@when('I register a skill "{name}" with tool refs in the registry')
def skill_registry_register_with_refs(context: Context, name: str) -> None:
"""Register a skill with tool refs in the registry."""
defn = _make_skill_definition(name, tool_refs=["local/tool-a", "local/tool-b"])
context.skill_registry.register(defn)
@when('I get skill "{name}" from the registry')
def skill_registry_get(context: Context, name: str) -> None:
"""Get a skill from the registry."""
context.registry_skill = context.skill_registry.get(name)
@when('I try to register a duplicate skill "{name}"')
def skill_registry_try_register_dup(context: Context, name: str) -> None:
"""Try to register a duplicate skill."""
context.skill_context_error = None
try:
defn = _make_skill_definition(name)
context.skill_registry.register(defn)
except SkillExecutionError as e:
context.skill_context_error = e
@when('I try to get skill "{name}" from the registry')
def skill_registry_try_get(context: Context, name: str) -> None:
"""Try to get a missing skill."""
context.skill_context_error = None
try:
context.skill_registry.get(name)
except SkillExecutionError as e:
context.skill_context_error = e
@when("I list all skills from the registry")
def skill_registry_list(context: Context) -> None:
"""List all skills from the registry."""
context.registry_skill_list = context.skill_registry.list_all()
@when('I unregister skill "{name}" from the registry')
def skill_registry_unregister(context: Context, name: str) -> None:
"""Unregister a skill from the registry."""
context.skill_registry.unregister(name)
@when('I try to unregister skill "{name}" from the registry')
def skill_registry_try_unregister(context: Context, name: str) -> None:
"""Try to unregister a missing skill."""
context.skill_context_error = None
try:
context.skill_registry.unregister(name)
except SkillExecutionError as e:
context.skill_context_error = e
@when('I resolve tools for "{name}" from the registry')
def skill_registry_resolve_tools(context: Context, name: str) -> None:
"""Resolve tools for a skill from the registry."""
context.resolved_tools = context.skill_registry.resolve_tools(name)
@then('the registry skill name should be "{expected}"')
def skill_registry_check_name(context: Context, expected: str) -> None:
"""Check registry skill name."""
assert context.registry_skill.skill.name == expected
@then("a skill_context validation error should be raised")
def skill_context_check_validation_error(context: Context) -> None:
"""Verify validation error was raised."""
assert context.skill_context_error is not None
assert context.skill_context_error.error_type.value == "validation_error"
@then("a skill_context not found error should be raised")
def skill_context_check_not_found_error(context: Context) -> None:
"""Verify not found error was raised."""
assert context.skill_context_error is not None
assert context.skill_context_error.error_type.value == "skill_not_found"
@then("the registry should have {count:d} skills")
def skill_registry_check_count(context: Context, count: int) -> None:
"""Check registry skill count."""
actual = len(context.registry_skill_list)
assert actual == count, f"Expected {count} skills, got {actual}"
@then('the registry skill list should contain "{name}"')
def skill_registry_check_list_contains(context: Context, name: str) -> None:
"""Check registry list contains a skill."""
names = [m.name for m in context.registry_skill_list]
assert name in names, f"Expected '{name}' in {names}"
@then("the resolved tools should have {count:d} entries")
def skill_registry_check_resolved_count(context: Context, count: int) -> None:
"""Check resolved tools count."""
actual = len(context.resolved_tools)
assert actual == count, f"Expected {count} entries, got {actual}"
@then('the resolved tools should contain "{name}"')
def skill_registry_check_resolved_contains(context: Context, name: str) -> None:
"""Check resolved tools contain a tool name."""
names = [e.name for e in context.resolved_tools]
assert name in names, f"Expected '{name}' in {names}"
# ---------------------------------------------------------------------------
# SkillRegistry Validation
# ---------------------------------------------------------------------------
@when("I create a skill_registry with a mock tool registry")
def skill_registry_create_with_mock(context: Context) -> None:
"""Create a SkillRegistry with a mock tool registry."""
mock_tool_reg = MagicMock()
mock_tool_reg.get_tool.return_value = None
context.skill_registry = SkillRegistry(tool_registry=mock_tool_reg)
context.skill_context_error = None
@when("I validate a skill with missing tool refs")
def skill_registry_validate_missing_refs(context: Context) -> None:
"""Validate a skill with tool refs that don't exist."""
defn = _make_skill_definition(
"local/validate-missing",
tool_refs=["local/nonexistent-tool"],
)
context.validation_errors = context.skill_registry.validate_skill(defn)
@when("I validate a skill with missing includes")
def skill_registry_validate_missing_includes(context: Context) -> None:
"""Validate a skill that includes a non-registered skill."""
from cleveragents.domain.models.core.skill import SkillInclude
skill = Skill(
name="local/validate-includes",
description="Skill with missing includes",
includes=[SkillInclude(name="local/not-registered")],
)
# Don't resolve tools through the resolver since the included
# skill doesn't exist — just build metadata from the skill alone
metadata = SkillMetadata(
name=skill.name,
description=skill.description,
)
defn = SkillDefinition(
skill=skill,
resolved_tools=[],
metadata=metadata,
)
context.validation_errors = context.skill_registry.validate_skill(defn)
@when("I validate a skill with no issues")
def skill_registry_validate_no_issues(context: Context) -> None:
"""Validate a skill with no issues."""
defn = _make_skill_definition("local/valid-skill")
context.validation_errors = context.skill_registry.validate_skill(defn)
@then('the validation errors should contain "{expected}"')
def skill_registry_check_validation_errors(context: Context, expected: str) -> None:
"""Check validation errors contain a message."""
assert any(expected in e for e in context.validation_errors), (
f"Expected error containing '{expected}' in {context.validation_errors}"
)
@then("the validation errors should be empty")
def skill_registry_check_validation_empty(context: Context) -> None:
"""Verify no validation errors."""
assert len(context.validation_errors) == 0, (
f"Expected empty errors, got {context.validation_errors}"
)
+236
View File
@@ -0,0 +1,236 @@
"""Step definitions for inline tool executor tests."""
from __future__ import annotations
from pathlib import Path
from behave import given, then, when
from behave.runner import Context
from cleveragents.domain.models.core.skill import SkillInlineTool
from cleveragents.domain.models.core.tool import ToolCapability, ToolSource
from cleveragents.skills.context import SkillContext
from cleveragents.skills.inline_executor import InlineToolExecutor
# ---------------------------------------------------------------------------
# Givens — Tool construction
# ---------------------------------------------------------------------------
@given('an inline tool with code that prints "{message}"')
def inline_tool_with_print(context: Context, message: str) -> None:
"""Create an inline tool that prints a message."""
context.inline_tool = SkillInlineTool(
description="Print message tool",
source=ToolSource.CUSTOM,
code=f'print("{message}")',
)
context.inline_executor = InlineToolExecutor()
@given("an inline tool with code that sleeps for 5 seconds")
def inline_tool_with_sleep(context: Context) -> None:
"""Create an inline tool that sleeps."""
context.inline_tool = SkillInlineTool(
description="Sleeping tool",
source=ToolSource.CUSTOM,
code="import time; time.sleep(5)",
)
context.inline_executor = InlineToolExecutor()
@given("an inline tool with code that prints 2000 bytes of output")
def inline_tool_with_large_output(context: Context) -> None:
"""Create an inline tool that produces large output."""
context.inline_tool = SkillInlineTool(
description="Large output tool",
source=ToolSource.CUSTOM,
code='print("X" * 2000)',
)
context.inline_executor = InlineToolExecutor()
@given('an inline tool with writes capability and code that prints "{message}"')
def inline_tool_with_writes(context: Context, message: str) -> None:
"""Create an inline tool with writes capability."""
context.inline_tool = SkillInlineTool(
description="Write-capable tool",
source=ToolSource.CUSTOM,
code=f'print("{message}")',
capability=ToolCapability(writes=True),
)
context.inline_executor = InlineToolExecutor()
@given("an inline tool with no code")
def inline_tool_no_code(context: Context) -> None:
"""Create an inline tool without code."""
context.inline_tool = SkillInlineTool(
description="No-code tool",
source=ToolSource.CUSTOM,
code=None,
)
context.inline_executor = InlineToolExecutor()
@given("an inline tool with unsupported source type")
def inline_tool_unsupported_source(context: Context) -> None:
"""Create an inline tool with unsupported source."""
context.inline_tool = SkillInlineTool(
description="Unsupported source tool",
source=ToolSource.BUILTIN,
code="print('hello')",
)
context.inline_executor = InlineToolExecutor()
@given("an inline tool with code that raises an exception")
def inline_tool_with_error(context: Context) -> None:
"""Create an inline tool that raises an exception."""
context.inline_tool = SkillInlineTool(
description="Error tool",
source=ToolSource.CUSTOM,
code='raise RuntimeError("intentional error")',
)
context.inline_executor = InlineToolExecutor()
# ---------------------------------------------------------------------------
# Givens — Context construction
# ---------------------------------------------------------------------------
@given('a writable skill context with sandbox "{sandbox}"')
def writable_skill_context(context: Context, sandbox: str) -> None:
"""Create a writable SkillContext."""
context.inline_context = SkillContext(
plan_id="plan-inline",
project_id="proj-inline",
sandbox_path=Path(sandbox),
read_only=False,
)
@given('a read-only skill context with sandbox "{sandbox}"')
def readonly_skill_context(context: Context, sandbox: str) -> None:
"""Create a read-only SkillContext."""
context.inline_context = SkillContext(
plan_id="plan-inline-ro",
project_id="proj-inline-ro",
sandbox_path=Path(sandbox),
read_only=True,
)
# ---------------------------------------------------------------------------
# Givens — Executor configuration
# ---------------------------------------------------------------------------
@given("an inline executor with max runtime {seconds:g} seconds")
def inline_executor_with_timeout(context: Context, seconds: float) -> None:
"""Create an executor with custom timeout."""
context.inline_executor = InlineToolExecutor(max_runtime_seconds=seconds)
@given("an inline executor with max output {nbytes:d} bytes")
def inline_executor_with_max_output(context: Context, nbytes: int) -> None:
"""Create an executor with custom output limit."""
context.inline_executor = InlineToolExecutor(max_output_bytes=nbytes)
# ---------------------------------------------------------------------------
# When — Execution
# ---------------------------------------------------------------------------
@when("I execute the inline tool")
def execute_inline_tool(context: Context) -> None:
"""Execute the inline tool with the configured executor and context."""
context.inline_result = context.inline_executor.execute(
tool=context.inline_tool,
context=context.inline_context,
input_data={},
)
@when("I validate the inline tool")
def validate_inline_tool(context: Context) -> None:
"""Validate the inline tool."""
context.inline_validation_errors = context.inline_executor.validate_tool(
context.inline_tool
)
# ---------------------------------------------------------------------------
# Then — Result assertions
# ---------------------------------------------------------------------------
@then("the inline result should be successful")
def inline_result_successful(context: Context) -> None:
"""Assert the inline result is successful."""
assert context.inline_result.success is True, (
f"Expected success but got error: {context.inline_result.error_message}"
)
@then("the inline result should not be successful")
def inline_result_not_successful(context: Context) -> None:
"""Assert the inline result is not successful."""
assert context.inline_result.success is False, "Expected failure but got success"
@then('the inline result output should contain "{expected}"')
def inline_result_output_contains(context: Context, expected: str) -> None:
"""Assert the inline result output contains expected text."""
output = str(context.inline_result.output or "")
assert expected in output, f"Expected '{expected}' in output '{output}'"
@then('the inline result error message should contain "{expected}"')
def inline_result_error_contains(context: Context, expected: str) -> None:
"""Assert the inline result error contains expected text."""
msg = context.inline_result.error_message or ""
assert expected in msg, f"Expected '{expected}' in error '{msg}'"
@then("the inline result should be truncated")
def inline_result_truncated(context: Context) -> None:
"""Assert the inline result output was truncated."""
assert context.inline_result.truncated is True, "Expected truncated=True"
# ---------------------------------------------------------------------------
# Then — Validation assertions
# ---------------------------------------------------------------------------
@then('the inline validation errors should contain "{expected}"')
def inline_validation_errors_contain(context: Context, expected: str) -> None:
"""Assert validation errors contain expected text."""
errors = context.inline_validation_errors
assert any(expected in e for e in errors), (
f"Expected error containing '{expected}' in {errors}"
)
# ---------------------------------------------------------------------------
# Then — Change tracker assertions
# ---------------------------------------------------------------------------
@then("the skill context change tracker should have {count:d} inline records")
def inline_tracker_count(context: Context, count: int) -> None:
"""Assert change tracker record count."""
actual = len(context.inline_context.change_tracker)
assert actual == count, f"Expected {count} records, got {actual}"
@then('the last inline tracker record tool_name should be "{expected}"')
def inline_tracker_last_tool(context: Context, expected: str) -> None:
"""Assert last tracker record tool_name."""
record = context.inline_context.change_tracker[-1]
assert record["tool_name"] == expected, (
f"Expected tool_name '{expected}', got '{record['tool_name']}'"
)
+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"
+59 -59
View File
@@ -3291,65 +3291,65 @@ 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 23) - 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 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`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master`
- [ ] Code [Jeff]: Implement SkillContext (plan/resource access, sandbox path, change tracker) and SkillRegistry in `src/cleveragents/skills/context.py`.
- [ ] Code [Jeff]: Wire SkillRegistry to Tool Registry for tool resolution and validation node inclusion.
- [ ] Code [Jeff]: Add context helpers for resolving bound resources and exposing plan metadata.
- [ ] Code [Jeff]: Include change tracking hooks so every skill execution registers a ToolInvocation in ChangeSet.
- [ ] Code [Jeff]: Ensure SkillContext enforces read-only plan restrictions (block write tools when plan is read-only).
- [ ] Docs [Jeff]: Add `docs/reference/skills_context.md` with context fields and helper methods.
- [ ] Tests (Behave) [Jeff]: Add `features/skill_context.feature` for sandboxed access and registry resolution.
- [ ] Tests (Robot) [Jeff]: Add `robot/skill_context.robot` for registry smoke tests.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/skill_context_bench.py` for registry resolution overhead.
- [ ] 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 context and registry"`
- [ ] 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 context, registry, and change tracking hooks.".
- [ ] **COMMIT (Owner: Jeff | Group: C3.inline | Branch: feature/m3-skill-protocol | Planned: Day 17 | Expected: Day 24) - Commit message: "feat(skill): add inline tool executor"**
- [ ] 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]: Implement inline tool execution with timeouts and restricted environment in `src/cleveragents/skills/inline_executor.py`.
- [ ] Code [Jeff]: Ensure inline tools conform to Tool Registry schema and return structured results.
- [ ] Code [Jeff]: Add safeguards for file/network access inside inline tools (local-only for MVP).
- [ ] Code [Jeff]: Add max runtime and max output size guards; return structured error on timeout/overflow.
- [ ] Docs [Jeff]: Add `docs/reference/skills_inline.md` with safety constraints.
- [ ] Tests (Behave) [Jeff]: Add `features/skill_inline.feature` for execution and timeout handling.
- [ ] Tests (Robot) [Jeff]: Add `robot/skill_inline.robot` for inline tool smoke tests.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/inline_tool_bench.py` for execution overhead.
- [ ] 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 inline tool executor"`
- [ ] 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 inline tool executor with safety constraints and tests.".
- [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.".
- [X] **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"**
- [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]: Implement SkillContext (plan/resource access, sandbox path, change tracker) and SkillRegistry in `src/cleveragents/skills/context.py`.
- [X] Code [Jeff]: Wire SkillRegistry to Tool Registry for tool resolution and validation node inclusion.
- [X] Code [Jeff]: Add context helpers for resolving bound resources and exposing plan metadata.
- [X] Code [Jeff]: Include change tracking hooks so every skill execution registers a ToolInvocation in ChangeSet.
- [X] Code [Jeff]: Ensure SkillContext enforces read-only plan restrictions (block write tools when plan is read-only).
- [X] Docs [Jeff]: Add `docs/reference/skills_context.md` with context fields and helper methods.
- [X] Tests (Behave) [Jeff]: Add `features/skill_context.feature` for sandboxed access and registry resolution.
- [X] Tests (Robot) [Jeff]: Add `robot/skill_context.robot` for registry smoke tests.
- [X] Tests (ASV) [Jeff]: Add `benchmarks/skill_context_bench.py` for registry resolution overhead.
- [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 context and registry"`
- [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 context, registry, and change tracking hooks.".
- [X] **COMMIT (Owner: Jeff | Group: C3.inline | Branch: feature/m3-skill-protocol | Planned: Day 17 | Expected: Day 24) - Commit message: "feat(skill): add inline tool executor"**
- [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]: Implement inline tool execution with timeouts and restricted environment in `src/cleveragents/skills/inline_executor.py`.
- [X] Code [Jeff]: Ensure inline tools conform to Tool Registry schema and return structured results.
- [X] Code [Jeff]: Add safeguards for file/network access inside inline tools (local-only for MVP).
- [X] Code [Jeff]: Add max runtime and max output size guards; return structured error on timeout/overflow.
- [X] Docs [Jeff]: Add `docs/reference/skills_inline.md` with safety constraints.
- [X] Tests (Behave) [Jeff]: Add `features/skill_inline.feature` for execution and timeout handling.
- [X] Tests (Robot) [Jeff]: Add `robot/skill_inline.robot` for inline tool smoke tests.
- [X] Tests (ASV) [Jeff]: Add `benchmarks/inline_tool_bench.py` for execution overhead.
- [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 inline tool executor"`
- [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 inline tool executor with safety constraints and tests.".
**Parallel Group C4: Built-in Skills [Jeff + Luis]** (depends on C3)
- [ ] **COMMIT (Owner: Jeff | Group: C4.file | Branch: feature/m3-skill-file-search | Planned: Day 18 | Expected: Day 25) - Commit message: "feat(skill): add file operation skills"**
+197
View File
@@ -0,0 +1,197 @@
"""Robot Framework helper for skill context and registry smoke tests.
Provides a CLI-style interface for Robot to invoke context creation,
resource resolution, change tracking, write guard, and registry operations.
Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_skill_context.py create-context
python robot/helper_skill_context.py resolve-resource
python robot/helper_skill_context.py track-invocation
python robot/helper_skill_context.py write-guard
python robot/helper_skill_context.py registry-crud
python robot/helper_skill_context.py registry-list
python robot/helper_skill_context.py registry-resolve
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
# 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.context import SkillContext, SkillExecutionError # noqa: E402
from cleveragents.skills.protocol import ( # noqa: E402
SkillDefinition,
SkillMetadata,
)
from cleveragents.skills.registry import SkillRegistry # noqa: E402
def _make_defn(name: str, **kw: Any) -> SkillDefinition:
"""Create a SkillDefinition with defaults."""
skill = Skill(name=name, description=f"Test {name}", **kw)
resolver = SkillResolver()
resolved = resolver.resolve_tools(skill, {})
meta = SkillMetadata.from_skill(skill, resolved=resolved)
return SkillDefinition(skill=skill, resolved_tools=resolved, metadata=meta)
def _cmd_create_context() -> int:
"""Test SkillContext creation."""
ctx = SkillContext(
plan_id="plan-robot",
project_id="proj-robot",
sandbox_path=Path("/tmp/robot-sandbox"),
)
if ctx.plan_id != "plan-robot":
print(f"skill-context-fail: unexpected plan_id {ctx.plan_id}")
return 1
if ctx.is_read_only():
print("skill-context-fail: should not be read-only")
return 1
print(f"skill-context-ok: plan={ctx.plan_id}")
return 0
def _cmd_resolve_resource() -> int:
"""Test resource resolution."""
ctx = SkillContext(
plan_id="plan-res",
project_id="proj-res",
sandbox_path=Path("/tmp/res-sandbox"),
resource_bindings={"git-checkout": "/repo/path"},
)
result = ctx.resolve_resource("git-checkout")
if result != "/repo/path":
print(f"skill-context-fail: unexpected resource {result}")
return 1
print(f"skill-context-resolve-ok: {result}")
return 0
def _cmd_track_invocation() -> int:
"""Test change tracking."""
ctx = SkillContext(
plan_id="plan-track",
project_id="proj-track",
sandbox_path=Path("/tmp/track-sandbox"),
)
ctx.register_tool_invocation("local/test-tool", {"a": 1}, {"b": 2}, 15.5)
if len(ctx.change_tracker) != 1:
print(f"skill-context-fail: expected 1 record, got {len(ctx.change_tracker)}")
return 1
if ctx.change_tracker[0]["tool_name"] != "local/test-tool":
print("skill-context-fail: wrong tool_name in tracker")
return 1
print("skill-context-track-ok")
return 0
def _cmd_write_guard() -> int:
"""Test write guard enforcement."""
ctx = SkillContext(
plan_id="plan-guard",
project_id="proj-guard",
sandbox_path=Path("/tmp/guard-sandbox"),
read_only=True,
)
try:
ctx.enforce_write_guard("local/write-tool")
print("skill-context-fail: expected SkillError")
return 1
except SkillExecutionError as e:
if e.error_type.value != "permission_denied":
print(f"skill-context-fail: unexpected error type {e.error_type}")
return 1
print("skill-context-guard-ok")
return 0
def _cmd_registry_crud() -> int:
"""Test registry register/get/unregister."""
reg = SkillRegistry()
defn = _make_defn("local/robot-reg")
reg.register(defn)
got = reg.get("local/robot-reg")
if got.skill.name != "local/robot-reg":
print(f"skill-registry-fail: unexpected name {got.skill.name}")
return 1
reg.unregister("local/robot-reg")
try:
reg.get("local/robot-reg")
print("skill-registry-fail: expected not-found error")
return 1
except SkillExecutionError:
pass
print("skill-registry-crud-ok")
return 0
def _cmd_registry_list() -> int:
"""Test registry list_all."""
reg = SkillRegistry()
reg.register(_make_defn("local/list-a"))
reg.register(_make_defn("local/list-b"))
items = reg.list_all()
if len(items) != 2:
print(f"skill-registry-fail: expected 2, got {len(items)}")
return 1
print(f"skill-registry-list-ok: {len(items)} skills")
return 0
def _cmd_registry_resolve() -> int:
"""Test registry tool resolution."""
reg = SkillRegistry()
defn = _make_defn("local/resolve-skill", tool_refs=["local/tool-x"])
reg.register(defn)
tools = reg.resolve_tools("local/resolve-skill")
if len(tools) != 1:
print(f"skill-registry-fail: expected 1 tool, got {len(tools)}")
return 1
print(f"skill-registry-resolve-ok: {len(tools)} tools")
return 0
_COMMANDS = {
"create-context": _cmd_create_context,
"resolve-resource": _cmd_resolve_resource,
"track-invocation": _cmd_track_invocation,
"write-guard": _cmd_write_guard,
"registry-crud": _cmd_registry_crud,
"registry-list": _cmd_registry_list,
"registry-resolve": _cmd_registry_resolve,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(
"Usage: helper_skill_context.py "
"<create-context|resolve-resource|track-invocation|write-guard"
"|registry-crud|registry-list|registry-resolve>"
)
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())
+182
View File
@@ -0,0 +1,182 @@
"""Robot Framework helper for inline tool executor smoke tests.
Provides a CLI-style interface for Robot to invoke inline tool
execution, validation, and write-guard checks.
Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_skill_inline.py execute-simple
python robot/helper_skill_inline.py execute-timeout
python robot/helper_skill_inline.py execute-truncate
python robot/helper_skill_inline.py validate-no-code
python robot/helper_skill_inline.py write-guard
"""
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
SkillInlineTool,
)
from cleveragents.domain.models.core.tool import ( # noqa: E402
ToolCapability,
ToolSource,
)
from cleveragents.skills.context import SkillContext # noqa: E402
from cleveragents.skills.inline_executor import InlineToolExecutor # noqa: E402
def _cmd_execute_simple() -> int:
"""Test simple inline tool execution."""
tool = SkillInlineTool(
description="Simple print",
source=ToolSource.CUSTOM,
code='print("hello-robot")',
)
ctx = SkillContext(
plan_id="plan-robot-inline",
project_id="proj-robot-inline",
sandbox_path=Path("/tmp/robot-inline-sandbox"),
)
executor = InlineToolExecutor()
result = executor.execute(tool, ctx, {})
if not result.success:
print(f"inline-fail: {result.error_message}")
return 1
if "hello-robot" not in str(result.output):
print(f"inline-fail: unexpected output {result.output}")
return 1
print("inline-execute-ok")
return 0
def _cmd_execute_timeout() -> int:
"""Test inline tool timeout."""
tool = SkillInlineTool(
description="Sleeping tool",
source=ToolSource.CUSTOM,
code="import time; time.sleep(5)",
)
ctx = SkillContext(
plan_id="plan-robot-timeout",
project_id="proj-robot-timeout",
sandbox_path=Path("/tmp/robot-timeout-sandbox"),
)
executor = InlineToolExecutor(max_runtime_seconds=0.1)
result = executor.execute(tool, ctx, {})
if result.success:
print("inline-fail: expected timeout failure")
return 1
if "timed out" not in (result.error_message or ""):
print(f"inline-fail: unexpected error: {result.error_message}")
return 1
print("inline-timeout-ok")
return 0
def _cmd_execute_truncate() -> int:
"""Test inline tool output truncation."""
tool = SkillInlineTool(
description="Large output tool",
source=ToolSource.CUSTOM,
code='print("X" * 2000)',
)
ctx = SkillContext(
plan_id="plan-robot-truncate",
project_id="proj-robot-truncate",
sandbox_path=Path("/tmp/robot-truncate-sandbox"),
)
executor = InlineToolExecutor(max_output_bytes=100)
result = executor.execute(tool, ctx, {})
if not result.success:
print(f"inline-fail: {result.error_message}")
return 1
if not result.truncated:
print("inline-fail: expected truncated=True")
return 1
print("inline-truncate-ok")
return 0
def _cmd_validate_no_code() -> int:
"""Test validation of tool without code."""
tool = SkillInlineTool(
description="No-code tool",
source=ToolSource.CUSTOM,
code=None,
)
executor = InlineToolExecutor()
errors = executor.validate_tool(tool)
if not errors:
print("inline-fail: expected validation errors")
return 1
if not any("non-empty code" in e for e in errors):
print(f"inline-fail: unexpected errors: {errors}")
return 1
print("inline-validate-ok")
return 0
def _cmd_write_guard() -> int:
"""Test write guard blocks execution in read-only context."""
tool = SkillInlineTool(
description="Write tool",
source=ToolSource.CUSTOM,
code='print("writing")',
capability=ToolCapability(writes=True),
)
ctx = SkillContext(
plan_id="plan-robot-guard",
project_id="proj-robot-guard",
sandbox_path=Path("/tmp/robot-guard-sandbox"),
read_only=True,
)
executor = InlineToolExecutor()
result = executor.execute(tool, ctx, {})
if result.success:
print("inline-fail: expected write guard to block")
return 1
if "denied" not in (result.error_message or "").lower():
print(f"inline-fail: unexpected error: {result.error_message}")
return 1
print("inline-guard-ok")
return 0
_COMMANDS = {
"execute-simple": _cmd_execute_simple,
"execute-timeout": _cmd_execute_timeout,
"execute-truncate": _cmd_execute_truncate,
"validate-no-code": _cmd_validate_no_code,
"write-guard": _cmd_write_guard,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(
"Usage: helper_skill_inline.py "
"<execute-simple|execute-timeout|execute-truncate"
"|validate-no-code|write-guard>"
)
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())
+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())
+65
View File
@@ -0,0 +1,65 @@
*** Settings ***
Documentation Smoke tests for Skill context and registry
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_skill_context.py
*** Test Cases ***
Create Skill Context
[Documentation] Create SkillContext with plan/project/sandbox info
${result}= Run Process ${PYTHON} ${HELPER} create-context cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-context-ok
Resolve Resource From Context
[Documentation] Resolve a bound resource from SkillContext
${result}= Run Process ${PYTHON} ${HELPER} resolve-resource cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-context-resolve-ok
Register Tool Invocation
[Documentation] Register a tool invocation in the change tracker
${result}= Run Process ${PYTHON} ${HELPER} track-invocation cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-context-track-ok
Enforce Write Guard
[Documentation] Enforce read-only write guard raises SkillError
${result}= Run Process ${PYTHON} ${HELPER} write-guard cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-context-guard-ok
Registry Register And Get
[Documentation] Register and retrieve a skill from the registry
${result}= Run Process ${PYTHON} ${HELPER} registry-crud cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-registry-crud-ok
Registry List All
[Documentation] List all skills from the registry
${result}= Run Process ${PYTHON} ${HELPER} registry-list cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-registry-list-ok
Registry Resolve Tools
[Documentation] Resolve tools from the registry
${result}= Run Process ${PYTHON} ${HELPER} registry-resolve cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-registry-resolve-ok
+49
View File
@@ -0,0 +1,49 @@
*** Settings ***
Documentation Smoke tests for inline tool executor
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_skill_inline.py
*** Test Cases ***
Execute Simple Inline Tool
[Documentation] Execute inline tool and capture output
${result}= Run Process ${PYTHON} ${HELPER} execute-simple cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} inline-execute-ok
Execute Inline Tool With Timeout
[Documentation] Inline tool times out with structured error
${result}= Run Process ${PYTHON} ${HELPER} execute-timeout cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} inline-timeout-ok
Execute Inline Tool With Truncation
[Documentation] Inline tool output is truncated at max size
${result}= Run Process ${PYTHON} ${HELPER} execute-truncate cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} inline-truncate-ok
Validate Inline Tool Missing Code
[Documentation] Validation detects missing code
${result}= Run Process ${PYTHON} ${HELPER} validate-no-code cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} inline-validate-ok
Write Guard Blocks Read Only
[Documentation] Write guard prevents execution in read-only context
${result}= Run Process ${PYTHON} ${HELPER} write-guard cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} inline-guard-ok
+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
+40
View File
@@ -0,0 +1,40 @@
"""Skill protocol, context, registry, and inline executor for CleverAgents v3.
This package defines the skill protocol types used for discovery,
composition, execution, and error reporting within the skill framework,
as well as the runtime context, skill registry, and inline tool executor.
"""
from cleveragents.skills.context import (
SkillContext,
SkillExecutionError,
ToolInvocationRecord,
)
from cleveragents.skills.inline_executor import (
InlineToolExecutor,
InlineToolResult,
)
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillErrorType,
SkillMetadata,
SkillResult,
map_tool_error,
)
from cleveragents.skills.registry import SkillRegistry
__all__ = [
"InlineToolExecutor",
"InlineToolResult",
"SkillContext",
"SkillDefinition",
"SkillError",
"SkillErrorType",
"SkillExecutionError",
"SkillMetadata",
"SkillRegistry",
"SkillResult",
"ToolInvocationRecord",
"map_tool_error",
]
+235
View File
@@ -0,0 +1,235 @@
"""Skill execution context for CleverAgents v3.
Provides the runtime context for skill execution, including plan and
project identifiers, sandbox path resolution, resource bindings, change
tracking, and read-only enforcement.
## Overview
- **SkillContext** -- Mutable runtime context that accumulates state
during skill execution. Carries plan/project identifiers, a sandbox
root ``Path``, bound resources, and a change tracker for recording
tool invocations.
## Read-Only Enforcement
When ``read_only=True``, the ``enforce_write_guard`` method raises a
``SkillExecutionError`` with error type ``PERMISSION_DENIED`` for any
tool that attempts to write. This is the primary mechanism for
enforcing plan-level read restrictions.
Based on ``docs/specification.md`` and ``implementation_plan.md`` task
C3.context.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from cleveragents.skills.protocol import SkillError, SkillErrorType
# ---------------------------------------------------------------------------
# Exception wrapper
# ---------------------------------------------------------------------------
class SkillExecutionError(Exception):
"""Exception raised for skill execution failures.
Wraps a ``SkillError`` payload so it can be raised and caught
while still carrying structured error data.
"""
def __init__(self, skill_error: SkillError) -> None:
self.skill_error = skill_error
super().__init__(skill_error.message)
@property
def error_type(self) -> SkillErrorType:
"""Return the error type from the wrapped error."""
return self.skill_error.error_type
@property
def tool_name(self) -> str | None:
"""Return the tool name from the wrapped error."""
return self.skill_error.tool_name
@property
def details(self) -> dict[str, Any]:
"""Return the details from the wrapped error."""
return self.skill_error.details
# ---------------------------------------------------------------------------
# Change tracker type alias
# ---------------------------------------------------------------------------
ToolInvocationRecord = dict[str, Any]
"""A single tool invocation record stored in the change tracker."""
# ---------------------------------------------------------------------------
# SkillContext
# ---------------------------------------------------------------------------
class SkillContext:
"""Runtime context for skill execution.
Accumulates state during a skill's lifecycle. **Not frozen** --
mutable by design so that tool invocations can be recorded and
resources can be resolved lazily.
Attributes:
plan_id: Identifier for the current plan.
project_id: Identifier for the current project.
sandbox_path: Root path to the execution sandbox.
change_tracker: List of recorded tool invocation records.
resource_bindings: Mapping of resource names to bound values.
read_only: Whether the context forbids write operations.
metadata: Arbitrary plan/project metadata dict.
"""
def __init__(
self,
plan_id: str,
project_id: str,
sandbox_path: Path,
*,
change_tracker: list[ToolInvocationRecord] | None = None,
resource_bindings: dict[str, Any] | None = None,
read_only: bool = False,
metadata: dict[str, Any] | None = None,
) -> None:
self.plan_id = plan_id
self.project_id = project_id
self.sandbox_path = sandbox_path
self.change_tracker: list[ToolInvocationRecord] = (
change_tracker if change_tracker is not None else []
)
self.resource_bindings: dict[str, Any] = (
resource_bindings if resource_bindings is not None else {}
)
self.read_only = read_only
self.metadata: dict[str, Any] = metadata if metadata is not None else {}
# -- Resource resolution ---------------------------------------------------
def resolve_resource(self, name: str) -> Any:
"""Look up a bound resource by name.
Args:
name: The resource binding key.
Returns:
The bound resource value.
Raises:
SkillExecutionError: With ``RESOLUTION_FAILURE`` if the
resource name is not found in the bindings.
"""
if name not in self.resource_bindings:
raise SkillExecutionError(
SkillError(
error_type=SkillErrorType.RESOLUTION_FAILURE,
message=f"Resource '{name}' not found in context bindings",
skill_name=f"context/{self.plan_id}",
details={"resource_name": name},
)
)
return self.resource_bindings[name]
# -- Plan metadata ---------------------------------------------------------
def get_plan_metadata(self) -> dict[str, Any]:
"""Return plan-related metadata dict.
Returns:
A dict containing ``plan_id``, ``project_id``, and any
additional metadata stored on the context.
"""
return {
"plan_id": self.plan_id,
"project_id": self.project_id,
**self.metadata,
}
# -- Sandbox path ----------------------------------------------------------
def get_sandbox_path(self) -> Path:
"""Return the sandbox root path.
Returns:
The ``Path`` to the sandbox root directory.
"""
return self.sandbox_path
# -- Read-only check -------------------------------------------------------
def is_read_only(self) -> bool:
"""Check whether this context forbids write operations.
Returns:
``True`` if the context is read-only.
"""
return self.read_only
# -- Change tracking -------------------------------------------------------
def register_tool_invocation(
self,
tool_name: str,
input_data: Any,
output_data: Any,
duration_ms: float,
) -> None:
"""Record a tool invocation in the change tracker.
Args:
tool_name: Name of the invoked tool.
input_data: Input data passed to the tool.
output_data: Output data returned by the tool.
duration_ms: Execution duration in milliseconds.
"""
record: ToolInvocationRecord = {
"tool_name": tool_name,
"input_data": input_data,
"output_data": output_data,
"duration_ms": duration_ms,
"plan_id": self.plan_id,
"project_id": self.project_id,
}
self.change_tracker.append(record)
# -- Write guard -----------------------------------------------------------
def enforce_write_guard(self, tool_name: str) -> None:
"""Raise ``SkillExecutionError(PERMISSION_DENIED)`` if read-only.
Should be called before any tool that performs write operations.
Args:
tool_name: The name of the tool attempting to write.
Raises:
SkillExecutionError: With ``PERMISSION_DENIED`` if the
context is read-only.
"""
if self.read_only:
raise SkillExecutionError(
SkillError(
error_type=SkillErrorType.PERMISSION_DENIED,
message=(
f"Write operation denied: tool '{tool_name}' cannot "
f"write in read-only context (plan={self.plan_id})"
),
skill_name=f"context/{self.plan_id}",
tool_name=tool_name,
details={
"plan_id": self.plan_id,
"project_id": self.project_id,
"read_only": True,
},
)
)
+402
View File
@@ -0,0 +1,402 @@
"""Inline tool executor for CleverAgents v3.
Provides safe execution of inline (anonymous) tool code within
skills. Inline tools are defined directly in skill YAML as Python
code strings and are executed in a restricted thread with timeout
and output-size guards.
## Safety Guards
- **Timeout**: Execution is bounded by ``max_runtime_seconds``
(default 30 s). A background thread runs the code; if it does
not complete in time the result reports a timeout error.
- **Output size**: Captured stdout/stderr is truncated at
``max_output_bytes`` (default 1 MiB). When truncation occurs the
``InlineToolResult.truncated`` flag is set.
- **Write guard**: If the inline tool's capability declares
``writes=True``, the executor calls
``SkillContext.enforce_write_guard`` before running, which raises
``SkillExecutionError(PERMISSION_DENIED)`` in a read-only
context.
- **Path restriction**: Any file-path values in *input_data* are
validated to reside within the sandbox root returned by
``SkillContext.get_sandbox_path()``.
- **Language check**: Only ``source=custom`` (Python) is supported
for MVP. Other source types are rejected by ``validate_tool``.
- **Network access**: For MVP this is documented as a local-only
restriction. No runtime enforcement is applied; the code runs
in the same process without network sandboxing.
Based on ``docs/specification.md`` and ``implementation_plan.md``
task C3.inline.
"""
from __future__ import annotations
import json
import os
import subprocess
import sys
import textwrap
import time
from pathlib import Path
from typing import Any
from pydantic import BaseModel, ConfigDict, Field
from cleveragents.domain.models.core.skill import SkillInlineTool
from cleveragents.domain.models.core.tool import ToolSource
from cleveragents.skills.context import SkillContext, SkillExecutionError
# ---------------------------------------------------------------------------
# InlineToolResult
# ---------------------------------------------------------------------------
class InlineToolResult(BaseModel):
"""Structured result of an inline tool execution.
Captures success/failure, captured output, timing, and whether
the output was truncated due to size limits.
"""
success: bool = Field(..., description="Whether the execution succeeded")
output: Any = Field(default=None, description="Captured stdout/stderr output")
error_message: str | None = Field(
default=None,
description="Error description on failure",
)
duration_ms: float = Field(
default=0.0,
ge=0.0,
description="Execution duration in milliseconds",
)
truncated: bool = Field(
default=False,
description="Whether output was truncated due to size limits",
)
model_config = ConfigDict(
frozen=True,
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# Supported languages
# ---------------------------------------------------------------------------
_SUPPORTED_SOURCES: frozenset[ToolSource] = frozenset({ToolSource.CUSTOM})
"""Tool sources that the inline executor can run (Python-only for MVP)."""
# ---------------------------------------------------------------------------
# InlineToolExecutor
# ---------------------------------------------------------------------------
class InlineToolExecutor:
"""Execute inline tool code with safety constraints.
Provides timeout enforcement, output-size truncation, write-guard
checks via the ``SkillContext``, and path-restriction validation.
Args:
max_runtime_seconds: Maximum wall-clock seconds for code
execution (default ``30.0``).
max_output_bytes: Maximum bytes of captured stdout/stderr
before truncation (default ``1_048_576`` = 1 MiB).
"""
def __init__(
self,
max_runtime_seconds: float = 30.0,
max_output_bytes: int = 1_048_576,
) -> None:
if max_runtime_seconds <= 0:
raise ValueError("max_runtime_seconds must be positive")
if max_output_bytes <= 0:
raise ValueError("max_output_bytes must be positive")
self._max_runtime_seconds = max_runtime_seconds
self._max_output_bytes = max_output_bytes
# -- Public API --------------------------------------------------------
@property
def max_runtime_seconds(self) -> float:
"""Return the configured maximum runtime in seconds."""
return self._max_runtime_seconds
@property
def max_output_bytes(self) -> int:
"""Return the configured maximum output size in bytes."""
return self._max_output_bytes
def execute(
self,
tool: SkillInlineTool,
context: SkillContext,
input_data: dict[str, Any],
) -> InlineToolResult:
"""Execute an inline tool's code in a guarded thread.
Steps:
1. Validate the tool.
2. Enforce write guard if the tool declares ``writes``.
3. Validate file paths in *input_data* against the sandbox.
4. Run the code string in a background thread with timeout.
5. Capture stdout/stderr, apply output-size truncation.
6. Record the invocation in the context's change tracker.
Args:
tool: The inline tool definition.
context: The skill execution context.
input_data: Input data passed to the tool code as a
local variable ``input_data``.
Returns:
An ``InlineToolResult`` with execution outcome.
"""
if tool is None:
raise ValueError("tool must not be None")
if context is None:
raise ValueError("context must not be None")
if input_data is None:
raise ValueError("input_data must not be None")
# Validate tool suitability
errors = self.validate_tool(tool)
if errors:
return InlineToolResult(
success=False,
error_message=f"Validation failed: {'; '.join(errors)}",
duration_ms=0.0,
)
# Enforce write guard for tools that write
if tool.capability is not None and tool.capability.writes:
try:
context.enforce_write_guard("inline_tool")
except SkillExecutionError as exc:
return InlineToolResult(
success=False,
error_message=str(exc),
duration_ms=0.0,
)
# Validate paths in input_data
path_error = self._validate_paths(input_data, context.get_sandbox_path())
if path_error is not None:
return InlineToolResult(
success=False,
error_message=path_error,
duration_ms=0.0,
)
# Execute with timeout
result = self._run_with_timeout(tool, input_data)
# Record invocation in change tracker
context.register_tool_invocation(
tool_name="inline_tool",
input_data=input_data,
output_data=result.output,
duration_ms=result.duration_ms,
)
return result
def validate_tool(self, tool: SkillInlineTool) -> list[str]:
"""Validate that an inline tool is suitable for execution.
Checks:
1. The tool has non-empty code.
2. The tool's source is a supported language/source type
(``custom`` / Python for MVP).
Args:
tool: The inline tool definition to validate.
Returns:
List of validation error messages. An empty list means
the tool is valid and ready for execution.
"""
if tool is None:
raise ValueError("tool must not be None")
errors: list[str] = []
if not tool.code:
errors.append("Inline tool must have non-empty code")
if tool.source not in _SUPPORTED_SOURCES:
errors.append(
f"Unsupported source '{tool.source.value}': "
f"only 'custom' (Python) is supported for MVP"
)
return errors
# -- Internal helpers --------------------------------------------------
def _validate_paths(
self,
input_data: dict[str, Any],
sandbox_path: Path,
) -> str | None:
"""Check that file-path values in *input_data* are within the sandbox.
Args:
input_data: The input data dict.
sandbox_path: The sandbox root path.
Returns:
An error message string if a path escapes the sandbox,
or ``None`` if all paths are valid.
"""
for key, value in input_data.items():
if not isinstance(value, str):
continue
# Heuristic: treat values that look like file paths
if key.endswith("_path") or key == "path" or key.endswith("_file"):
try:
resolved = Path(value).resolve()
sandbox_resolved = sandbox_path.resolve()
if not str(resolved).startswith(str(sandbox_resolved)):
return (
f"Path '{value}' for key '{key}' escapes sandbox "
f"root '{sandbox_path}'"
)
except (OSError, ValueError):
return f"Invalid path '{value}' for key '{key}'"
return None
def _run_with_timeout(
self,
tool: SkillInlineTool,
input_data: dict[str, Any],
) -> InlineToolResult:
"""Run tool code in a subprocess with timeout.
Launches a child Python process that executes the tool's code
string. The ``input_data`` dict is serialised as JSON and
injected into the child's namespace. Stdout and stderr are
captured and combined.
Using a subprocess (rather than an in-process thread) ensures:
- Clean timeout via ``subprocess.Popen.kill()``
- No lingering daemon threads that confuse coverage tooling
- Process-level isolation for the executed code
Applies output-size truncation when the combined output exceeds
``max_output_bytes``.
Args:
tool: The inline tool whose ``code`` to execute.
input_data: Passed into the execution namespace as
``input_data``.
Returns:
An ``InlineToolResult`` with captured output and timing.
"""
# Build a small wrapper script that deserialises input_data and
# execs the tool code, emitting a JSON result envelope on fd 3
# (or just use stdout/stderr capture for simplicity).
wrapper = textwrap.dedent("""\
import json, sys
input_data = json.loads(sys.argv[1])
try:
exec(sys.argv[2])
print("__INLINE_OK__")
except Exception as _exc:
print(f"__INLINE_ERR__{_exc}", file=sys.stderr)
sys.exit(1)
""")
# Build a clean environment for the subprocess.
# Remove coverage-related env vars to prevent the child from
# inheriting coverage instrumentation (which can cause hangs).
child_env = {
k: v for k, v in os.environ.items() if not k.startswith("COVERAGE")
}
start = time.monotonic()
try:
proc = subprocess.Popen(
[
sys.executable,
"-I",
"-c",
wrapper,
json.dumps(input_data),
tool.code or "",
],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=child_env,
)
try:
stdout_bytes, stderr_bytes = proc.communicate(
timeout=self._max_runtime_seconds,
)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
elapsed_ms = (time.monotonic() - start) * 1000.0
return InlineToolResult(
success=False,
error_message=(
f"Execution timed out after {self._max_runtime_seconds}s"
),
duration_ms=elapsed_ms,
)
except OSError as exc:
elapsed_ms = (time.monotonic() - start) * 1000.0
return InlineToolResult(
success=False,
error_message=f"Failed to start subprocess: {exc}",
duration_ms=elapsed_ms,
)
elapsed_ms = (time.monotonic() - start) * 1000.0
# Decode and combine output
raw_stdout = stdout_bytes.decode("utf-8", errors="replace")
raw_stderr = stderr_bytes.decode("utf-8", errors="replace")
raw_output = raw_stdout + raw_stderr
# Check for truncation
truncated = False
if len(raw_output.encode("utf-8", errors="replace")) > self._max_output_bytes:
encoded = raw_output.encode("utf-8", errors="replace")
raw_output = encoded[: self._max_output_bytes].decode(
"utf-8", errors="replace"
)
truncated = True
# Parse success/failure from the wrapper's sentinel
if proc.returncode == 0 and "__INLINE_OK__" in raw_stdout:
# Remove the sentinel from displayed output
clean_output = raw_stdout.replace("__INLINE_OK__\n", "").replace(
"__INLINE_OK__", ""
)
return InlineToolResult(
success=True,
output=clean_output + raw_stderr,
duration_ms=elapsed_ms,
truncated=truncated,
)
# Execution error — extract message from stderr sentinel
error_msg = raw_stderr
if "__INLINE_ERR__" in raw_stderr:
error_msg = raw_stderr.split("__INLINE_ERR__", 1)[1].strip()
return InlineToolResult(
success=False,
output=raw_stdout,
error_message=error_msg or "Unknown execution error",
duration_ms=elapsed_ms,
truncated=truncated,
)
+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
+215
View File
@@ -0,0 +1,215 @@
"""Skill registry for CleverAgents v3.
Manages registration, lookup, and validation of skill definitions.
Resolves tool references through an optional Tool Registry integration.
## Overview
- **SkillRegistry** -- In-memory registry of ``SkillDefinition`` instances,
keyed by skill name. Provides CRUD operations, metadata listing,
tool resolution, and definition validation.
## Tool Resolution
The ``resolve_tools`` method delegates to the skill's own
``SkillResolver`` to flatten tool references. When a ``_tool_registry``
reference is configured, tool-ref names can be validated against the
Tool Registry to ensure they exist.
## Validation
The ``validate_skill`` method checks that all tool references in a skill
definition actually exist (either in the tool registry or as inline
tools), and reports any discrepancies as a list of error messages.
Based on ``docs/specification.md`` and ``implementation_plan.md`` task
C3.context.
"""
from __future__ import annotations
from typing import Any
from cleveragents.domain.models.core.skill import (
ResolvedToolEntry,
SkillResolver,
)
from cleveragents.skills.context import SkillExecutionError
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
SkillErrorType,
SkillMetadata,
)
# ---------------------------------------------------------------------------
# SkillRegistry
# ---------------------------------------------------------------------------
class SkillRegistry:
"""In-memory registry for skill definitions.
Provides registration, lookup, listing, tool resolution, and
validation against an optional tool registry.
Attributes:
_skills: Internal mapping of skill names to definitions.
_tool_registry: Optional reference to a Tool Registry service
for tool-ref validation.
"""
def __init__(
self,
tool_registry: Any | None = None,
) -> None:
self._skills: dict[str, SkillDefinition] = {}
self._tool_registry: Any | None = tool_registry
# -- Registration ----------------------------------------------------------
def register(self, skill: SkillDefinition) -> None:
"""Register a skill definition.
Args:
skill: The ``SkillDefinition`` to register.
Raises:
SkillExecutionError: With ``VALIDATION_ERROR`` if a skill
with the same name is already registered.
"""
name = skill.skill.name
if name in self._skills:
raise SkillExecutionError(
SkillError(
error_type=SkillErrorType.VALIDATION_ERROR,
message=f"Skill '{name}' is already registered",
skill_name=name,
details={"existing_skill": name},
)
)
self._skills[name] = skill
def unregister(self, name: str) -> None:
"""Remove a skill definition from the registry.
Args:
name: The namespaced skill name to remove.
Raises:
SkillExecutionError: With ``SKILL_NOT_FOUND`` if the skill
is not registered.
"""
if name not in self._skills:
raise SkillExecutionError(
SkillError(
error_type=SkillErrorType.SKILL_NOT_FOUND,
message=f"Skill '{name}' is not registered",
skill_name=name,
)
)
del self._skills[name]
# -- Lookup ----------------------------------------------------------------
def get(self, name: str) -> SkillDefinition:
"""Look up a registered skill by name.
Args:
name: The namespaced skill name.
Returns:
The registered ``SkillDefinition``.
Raises:
SkillExecutionError: With ``SKILL_NOT_FOUND`` if the skill
is not registered.
"""
if name not in self._skills:
raise SkillExecutionError(
SkillError(
error_type=SkillErrorType.SKILL_NOT_FOUND,
message=f"Skill '{name}' not found in registry",
skill_name=name,
)
)
return self._skills[name]
# -- Listing ---------------------------------------------------------------
def list_all(self) -> list[SkillMetadata]:
"""Return metadata for all registered skills.
Returns:
List of ``SkillMetadata`` instances, one per registered
skill, sorted by name.
"""
return [
defn.metadata
for defn in sorted(self._skills.values(), key=lambda d: d.skill.name)
]
# -- Tool resolution -------------------------------------------------------
def resolve_tools(self, skill_name: str) -> list[ResolvedToolEntry]:
"""Resolve tool references for a registered skill.
Uses the ``SkillResolver`` to flatten the skill's tool set.
Other registered skills are available for include resolution.
Args:
skill_name: The namespaced skill name.
Returns:
Ordered list of ``ResolvedToolEntry`` instances.
Raises:
SkillExecutionError: With ``SKILL_NOT_FOUND`` if the skill
is not registered.
"""
defn = self.get(skill_name)
resolver = SkillResolver()
# Build a lookup of all registered skills for include resolution
skill_lookup = {name: d.skill for name, d in self._skills.items()}
return resolver.resolve_tools(defn.skill, skill_lookup)
# -- Validation ------------------------------------------------------------
def validate_skill(self, skill: SkillDefinition) -> list[str]:
"""Validate a skill definition's tool references.
Checks that:
1. All tool_refs refer to tools that exist in the tool registry
(if a tool registry is configured).
2. The skill's input/output schemas are well-formed.
3. Inline tools have required fields.
Args:
skill: The ``SkillDefinition`` to validate.
Returns:
List of validation error messages. Empty list means
the skill is valid.
"""
errors: list[str] = []
# Validate tool_refs against tool registry
for ref in skill.skill.tool_refs:
if self._tool_registry is not None:
tool = self._tool_registry.get_tool(ref)
if tool is None:
errors.append(f"Tool reference '{ref}' not found in tool registry")
# Validate inline tools have descriptions
for idx, inline in enumerate(skill.skill.anonymous_tools):
if not inline.description:
errors.append(f"Inline tool at index {idx} is missing a description")
# Validate includes reference known skills
for include in skill.skill.includes:
if include.name not in self._skills:
errors.append(f"Included skill '{include.name}' is not registered")
return errors
+5
View File
@@ -128,3 +128,8 @@ get_parents # noqa: B018, F821
auto_discover_children # noqa: B018, F821
_get_ancestors # noqa: B018, F821
_build_cycle_path # noqa: B018, F821
# Inline tool executor — public API
InlineToolExecutor # noqa: B018, F821
InlineToolResult # noqa: B018, F821
_SUPPORTED_SOURCES # noqa: B018, F821