HAL9000/temp
0
0
Fork 0
forked from HAL9000/cleveragents-core
Code Pull Requests Activity

feat(skill): add skill context and registry

Browse Source
This commit is contained in:
Jeffrey Phillips Freeman
2026-02-19 02:15:56 +00:00
parent 6e440ba4db
commit 94a78b3311
10 changed files with 1640 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/skill_context_bench.py
+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)
docs/reference/skills_context.md
+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).
features/skill_context.feature
+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
features/steps/skill_context_steps.py
+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}"
)
implementation_plan.md
+20 -20
View File
@@ -3311,26 +3311,26 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [X] Git [Jeff]: `git commit -m "feat(skill): add skill protocol and metadata"`
- [X] Git [Jeff]: `git push -u origin feature/m3-skill-protocol`
- [X] Forgejo PR [Jeff]: Open PR from `feature/m3-skill-protocol` to `master` with description "Add skill protocol, metadata models, and tests.".
- [ ] **COMMIT (Owner: Jeff | Group: C3.context | Branch: feature/m3-skill-protocol | Planned: Day 17 | Expected: Day 24) - Commit message: "feat(skill): add skill context and registry"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-skill-protocol`
- [ ] 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.".
- [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.".
- [ ] **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`
robot/helper_skill_context.py
+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())
robot/skill_context.robot
+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
src/cleveragents/skills/__init__.py
+13 -2
View File
@@ -1,9 +1,15 @@
"""Skill protocol and metadata for CleverAgents v3.
"""Skill protocol, context, and registry for CleverAgents v3.
This package defines the skill protocol types used for discovery,
composition, execution, and error reporting within the skill framework.
composition, execution, and error reporting within the skill framework,
as well as the runtime context and skill registry.
"""
from cleveragents.skills.context import (
SkillContext,
SkillExecutionError,
ToolInvocationRecord,
)
from cleveragents.skills.protocol import (
SkillDefinition,
SkillError,
@@ -12,12 +18,17 @@ from cleveragents.skills.protocol import (
SkillResult,
map_tool_error,
)
from cleveragents.skills.registry import SkillRegistry
__all__ = [
"SkillContext",
"SkillDefinition",
"SkillError",
"SkillErrorType",
"SkillExecutionError",
"SkillMetadata",
"SkillRegistry",
"SkillResult",
"ToolInvocationRecord",
"map_tool_error",
]
src/cleveragents/skills/context.py
+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,
},
)
)
src/cleveragents/skills/registry.py
+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