feat(tool): add tool and validation domain models #59

Merged
freemo merged 1 commits from feature/m3-tool-domain into master 2026-02-14 03:44:52 +00:00
13 changed files with 2616 additions and 0 deletions
+186
View File
@@ -0,0 +1,186 @@
"""ASV benchmarks for Tool and Validation domain model validation and serialization.
Measures the performance of:
- Tool model construction (Pydantic validation)
- Tool.as_cli_dict() serialization
- Tool.from_config() factory
- Validation model construction with forced constraints
- ResourceSlot construction
- ToolCapability construction
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.domain.models.core.tool import (
BindingMode,
ResourceAccessMode,
ResourceSlot,
Tool,
ToolCapability,
ToolLifecycle,
ToolSource,
ToolType,
Validation,
ValidationMode,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.domain.models.core.tool import (
BindingMode,
ResourceAccessMode,
ResourceSlot,
Tool,
ToolCapability,
ToolLifecycle,
ToolSource,
ToolType,
Validation,
ValidationMode,
)
def _make_tool() -> Tool:
"""Create a fully-populated Tool for benchmarking."""
return Tool(
name="bench/tool-test",
description="Benchmark tool for validation and serialization",
source=ToolSource.CUSTOM,
code="return {'result': True}",
input_schema={"type": "object", "properties": {"file": {"type": "string"}}},
output_schema={"type": "object", "properties": {"result": {"type": "boolean"}}},
capability=ToolCapability(
writes=True,
write_scope="file",
checkpointable=True,
side_effects=["filesystem"],
idempotent=False,
),
resource_slots=[
ResourceSlot(
name="repo",
resource_type="git-checkout",
access=ResourceAccessMode.READ_ONLY,
binding=BindingMode.CONTEXTUAL,
),
],
lifecycle=ToolLifecycle(
discover="bench.discover",
activate="bench.activate",
),
timeout=120,
)
def _make_validation() -> Validation:
"""Create a fully-populated Validation for benchmarking."""
return Validation(
name="bench/val-test",
description="Benchmark validation",
source=ToolSource.WRAPPED,
wraps="bench/tool-test",
transform="def transform(r): return {'passed': True}",
argument_mapping={"strict": True},
mode=ValidationMode.REQUIRED,
)
class ToolValidationSuite:
"""Benchmark Tool model construction (Pydantic validation)."""
def time_tool_construction(self) -> None:
"""Benchmark creating a fully-populated Tool object."""
_make_tool()
def time_tool_minimal_construction(self) -> None:
"""Benchmark creating a minimal Tool object."""
Tool(
name="bench/minimal",
description="Minimal tool",
source=ToolSource.BUILTIN,
)
def time_validation_construction(self) -> None:
"""Benchmark creating a Validation with forced constraints."""
_make_validation()
def time_resource_slot_construction(self) -> None:
"""Benchmark ResourceSlot construction."""
ResourceSlot(
name="repo",
resource_type="git-checkout",
access=ResourceAccessMode.READ_ONLY,
binding=BindingMode.CONTEXTUAL,
)
def time_capability_construction(self) -> None:
"""Benchmark ToolCapability construction."""
ToolCapability(
read_only=True,
writes=False,
checkpointable=False,
idempotent=True,
)
class ToolSerializationSuite:
"""Benchmark Tool serialization methods."""
def setup(self) -> None:
"""Create objects for serialization benchmarks."""
self.tool = _make_tool()
self.validation = _make_validation()
def time_tool_as_cli_dict(self) -> None:
"""Benchmark Tool.as_cli_dict() ordered dict generation."""
self.tool.as_cli_dict()
def time_validation_as_cli_dict(self) -> None:
"""Benchmark Validation.as_cli_dict() ordered dict generation."""
self.validation.as_cli_dict()
def time_tool_model_dump(self) -> None:
"""Benchmark Pydantic model_dump() serialization."""
self.tool.model_dump()
def time_tool_model_dump_json(self) -> None:
"""Benchmark Pydantic model_dump_json() JSON serialization."""
self.tool.model_dump_json()
class ToolFromConfigSuite:
"""Benchmark Tool.from_config() factory."""
def setup(self) -> None:
"""Prepare config dicts for benchmarks."""
self.tool_config = {
"name": "bench/from-config",
"description": "Config benchmark",
"source": "builtin",
"capability": {"read_only": True},
"resource_slots": [
{
"name": "repo",
"resource_type": "git-checkout",
"access": "read_only",
},
],
}
self.validation_config = {
"name": "bench/val-config",
"description": "Config benchmark validation",
"wraps": "bench/tool-test",
"transform": "def transform(r): return r",
"mode": "required",
}
def time_tool_from_config(self) -> None:
"""Benchmark Tool.from_config()."""
Tool.from_config(self.tool_config)
def time_validation_from_config(self) -> None:
"""Benchmark Validation.from_config()."""
Validation.from_config(self.validation_config)
+100
View File
@@ -0,0 +1,100 @@
# Tool Domain Model
The `Tool` is the atomic unit of execution in CleverAgents -- a namespaced,
independently registered callable operation.
## Tool Name
Tools use a `namespace/short_name` naming pattern:
```
local/line-counter
devops/docker-build
```
The name is the unique identifier. Tools and Validations share the same
namespace; no collisions are allowed.
## Source Types
| Source | Description | Required Fields |
|---------------|------------------------------------------------|---------------------------|
| `mcp` | Delegates to an MCP server | `mcp_server`, `mcp_tool_name` |
| `agent_skill` | Agent Skills Standard folder | `agent_skill_path` |
| `builtin` | Provided by the runtime | (none) |
| `custom` | Inline Python code | `code` |
| `wrapped` | Validation-only; wraps an existing tool | (set via Validation) |
## ToolCapability
Describes what a tool can and cannot do:
| Field | Type | Default | Description |
|--------------------------|----------------|---------|------------------------------------------|
| `read_only` | `bool` | `False` | Tool only reads, never writes |
| `writes` | `bool` | `False` | Tool can write to resources |
| `write_scope` | `str \| None` | `None` | Scope of writes |
| `checkpointable` | `bool` | `False` | Tool supports checkpoint/rollback |
| `checkpoint_scope` | `CheckpointScope \| None` | `None` | Granularity of checkpoint support |
| `side_effects` | `list[str]` | `[]` | Known side effects |
| `idempotent` | `bool` | `False` | Safe to re-run |
| `unsafe` | `bool` | `False` | Requires extra safety checks |
| `human_approval_required`| `bool` | `False` | Requires human approval |
| `cost_profile` | `str \| None` | `None` | Named cost profile |
**Constraint**: If `read_only=True`, then `writes` and `checkpointable` must be `False`.
## ResourceSlot
Declares a resource binding for a tool:
| Field | Type | Default | Description |
|-------------------|---------------------|---------------|-------------------------------------|
| `name` | `str` | (required) | Slot name (valid Python identifier) |
| `resource_type` | `str` | (required) | Resource type name |
| `access` | `ResourceAccessMode`| (required) | `read_only` or `read_write` |
| `description` | `str` | `""` | Human-readable description |
| `binding` | `BindingMode` | `contextual` | How the resource is resolved |
| `static_resource` | `str \| None` | `None` | Resource name (required for static) |
| `required` | `bool` | `True` | Whether slot must be filled |
**Constraint**: `static_resource` required when `binding=static`; forbidden otherwise.
## ToolLifecycle
Optional lifecycle hooks:
| Field | Type | Description |
|--------------|---------------|---------------------------|
| `discover` | `str \| None` | Hook for tool discovery |
| `activate` | `str \| None` | Hook for tool activation |
| `deactivate` | `str \| None` | Hook for tool deactivation|
## Tool Fields
| Field | Type | Default | Description |
|--------------------|-------------------------|------------------|----------------------------------------|
| `name` | `str` | (required) | `namespace/short_name` format |
| `description` | `str` | (required) | Short description |
| `source` | `ToolSource` | (required) | Implementation source |
| `tool_type` | `ToolType` | `tool` | Registry discriminator |
| `code` | `str \| None` | `None` | Inline Python (custom source) |
| `mcp_server` | `str \| None` | `None` | MCP server name |
| `mcp_tool_name` | `str \| None` | `None` | MCP tool name |
| `agent_skill_path` | `str \| None` | `None` | Agent Skills path |
| `input_schema` | `dict \| None` | `None` | JSON Schema for inputs |
| `output_schema` | `dict \| None` | `None` | JSON Schema for outputs |
| `capability` | `ToolCapability` | `ToolCapability()`| Capability metadata |
| `resource_slots` | `list[ResourceSlot]` | `[]` | Resource bindings |
| `lifecycle` | `ToolLifecycle \| None` | `None` | Lifecycle hooks |
| `timeout` | `int` | `300` | Execution timeout (seconds, >= 1) |
## Class Methods
- `Tool.from_config(config: dict)` -- Create from a YAML config dict
- `tool.as_cli_dict()` -- Stable ordered dict for CLI rendering
## Properties
- `tool.namespace` -- Extracted namespace from name
- `tool.short_name` -- Extracted short name from name
+57
View File
@@ -0,0 +1,57 @@
# Validation Domain Model
A `Validation` extends `Tool` with pass/fail semantics. Validations are
always read-only (`writes=False`, `checkpointable=False`, `read_only=True`).
## Validation-Specific Fields
| Field | Type | Default | Description |
|--------------------|-------------------------|-------------|------------------------------------------|
| `mode` | `ValidationMode` | `required` | Whether failure blocks execution |
| `wraps` | `str \| None` | `None` | Name of existing tool to wrap |
| `transform` | `str \| None` | `None` | Python code to transform wrapped output |
| `argument_mapping` | `dict \| None` | `None` | Args for wrapped tool (only with wraps) |
All fields from `Tool` are also available.
## Forced Constraints
When constructing a Validation, the following are always enforced:
- `tool_type` is forced to `ToolType.VALIDATION`
- `capability.read_only` is forced to `True`
- `capability.writes` is forced to `False`
- `capability.checkpointable` is forced to `False`
## Wraps Semantics
A Validation may wrap an existing Tool to reuse its implementation:
- When `wraps` is set, `source` must be `wrapped`
- When `wraps` is set, `code` must not be set (mutually exclusive)
- When `wraps` is set, `transform` is required
- `argument_mapping` is only valid when `wraps` is set
## Modes
| Mode | Behavior |
|-----------------|-------------------------------------------------|
| `required` | Failure blocks the operation |
| `informational` | Failure is reported but does not block |
## Class Methods
- `Validation.from_config(config: dict)` -- Create from a YAML config dict
(auto-detects `source=wrapped` when `wraps` is present)
- `validation.as_cli_dict()` -- Stable ordered dict with validation fields
## CLI Management
```
agents validation add --config <file> [--update]
agents validation attach [--project <project>|--plan <plan_id>] <resource> <validation>
agents validation detach [--yes] <attachment_id>
```
Validations are listed and inspected via standard `agents tool` commands
with `--type validation` filter.
+115
View File
@@ -0,0 +1,115 @@
# Tool YAML Schema
# Defines the expected structure for tool configuration files.
# See docs/specification.md for full details.
type: object
required:
- name
- description
- source
properties:
name:
type: string
pattern: "^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$"
description: "Namespaced tool name (namespace/short_name)"
description:
type: string
minLength: 1
description: "Short description of what the tool does"
source:
type: string
enum: [mcp, agent_skill, builtin, custom]
description: "Source of the tool implementation"
tool_type:
type: string
enum: [tool, validation]
default: tool
description: "Discriminator for the tool registry"
code:
type: string
description: "Inline Python code (required when source=custom)"
mcp_server:
type: string
description: "MCP server name (required when source=mcp)"
mcp_tool_name:
type: string
description: "Tool name on MCP server (required when source=mcp)"
agent_skill_path:
type: string
description: "Path to Agent Skills Standard folder (required when source=agent_skill)"
input_schema:
type: object
description: "JSON Schema for tool inputs"
output_schema:
type: object
description: "JSON Schema for tool outputs"
capability:
type: object
properties:
read_only:
type: boolean
default: false
writes:
type: boolean
default: false
write_scope:
type: string
checkpointable:
type: boolean
default: false
checkpoint_scope:
type: string
enum: [file, transaction, snapshot, composite]
side_effects:
type: array
items:
type: string
idempotent:
type: boolean
default: false
unsafe:
type: boolean
default: false
human_approval_required:
type: boolean
default: false
cost_profile:
type: string
resource_slots:
type: array
items:
type: object
required: [name, resource_type, access]
properties:
name:
type: string
resource_type:
type: string
access:
type: string
enum: [read_only, read_write]
description:
type: string
binding:
type: string
enum: [contextual, static, parameter]
default: contextual
static_resource:
type: string
required:
type: boolean
default: true
lifecycle:
type: object
properties:
discover:
type: string
activate:
type: string
deactivate:
type: string
timeout:
type: integer
minimum: 1
default: 300
description: "Execution timeout in seconds"
+94
View File
@@ -0,0 +1,94 @@
# Validation YAML Schema
# Extends the Tool schema with validation-specific fields.
# See docs/specification.md for full details.
type: object
required:
- name
- description
properties:
name:
type: string
pattern: "^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$"
description: "Namespaced validation name (namespace/short_name)"
description:
type: string
minLength: 1
description: "Short description of what the validation checks"
source:
type: string
enum: [mcp, agent_skill, builtin, custom, wrapped]
description: "Source of the validation implementation (inferred as 'wrapped' when wraps is set)"
mode:
type: string
enum: [required, informational]
default: required
description: "Whether failure blocks execution"
wraps:
type: string
description: "Name of existing tool this validation wraps"
transform:
type: string
description: "Python function code to transform wrapped tool output (required when wraps is set)"
argument_mapping:
type: object
description: "Argument mapping for the wrapped tool (only valid when wraps is set)"
code:
type: string
description: "Inline Python code (required when source=custom, mutually exclusive with wraps)"
mcp_server:
type: string
description: "MCP server name (required when source=mcp)"
mcp_tool_name:
type: string
description: "Tool name on MCP server (required when source=mcp)"
agent_skill_path:
type: string
description: "Path to Agent Skills Standard folder (required when source=agent_skill)"
input_schema:
type: object
description: "JSON Schema for validation inputs"
output_schema:
type: object
description: "JSON Schema for validation outputs"
capability:
type: object
description: "Capability metadata (read_only, writes, checkpointable forced for validations)"
resource_slots:
type: array
items:
type: object
required: [name, resource_type, access]
properties:
name:
type: string
resource_type:
type: string
access:
type: string
enum: [read_only, read_write]
description:
type: string
binding:
type: string
enum: [contextual, static, parameter]
default: contextual
static_resource:
type: string
required:
type: boolean
default: true
lifecycle:
type: object
properties:
discover:
type: string
activate:
type: string
deactivate:
type: string
timeout:
type: integer
minimum: 1
default: 300
description: "Execution timeout in seconds"
+40
View File
@@ -0,0 +1,40 @@
# Example: Custom Python tool
# A simple custom tool that runs inline Python code.
name: local/line-counter
description: Count lines in a file
source: custom
code: |
import pathlib
path = pathlib.Path(inputs["file_path"])
count = len(path.read_text().splitlines())
return {"line_count": count}
input_schema:
type: object
required:
- file_path
properties:
file_path:
type: string
description: Path to the file to count lines in
output_schema:
type: object
properties:
line_count:
type: integer
capability:
read_only: true
writes: false
idempotent: true
resource_slots:
- name: target_file
resource_type: fs-mount
access: read_only
description: File system containing the target file
binding: contextual
timeout: 60
+53
View File
@@ -0,0 +1,53 @@
# Example: MCP server tool
# Delegates execution to a tool on an MCP server.
name: devops/docker-build
description: Build a Docker image via the DevOps MCP server
source: mcp
mcp_server: devops-mcp
mcp_tool_name: docker_build
input_schema:
type: object
required:
- context_dir
- image_tag
properties:
context_dir:
type: string
description: Docker build context directory
image_tag:
type: string
description: Tag for the built image
dockerfile:
type: string
description: Path to Dockerfile (relative to context_dir)
default: Dockerfile
output_schema:
type: object
properties:
image_id:
type: string
build_log:
type: string
capability:
read_only: false
writes: true
write_scope: container-registry
checkpointable: false
side_effects:
- network
- container-registry
unsafe: false
human_approval_required: false
resource_slots:
- name: source_repo
resource_type: git-checkout
access: read_only
description: Source code repository for the Docker build
binding: contextual
timeout: 600
@@ -0,0 +1,39 @@
# Example: Required validation (standalone)
# A custom validation that checks test coverage meets a threshold.
name: qa/coverage-check
description: Verify test coverage meets the required threshold
source: custom
mode: required
code: |
import json
report = json.loads(inputs["coverage_json"])
total = report.get("totals", {}).get("percent_covered", 0)
passed = total >= inputs.get("threshold", 80)
return {
"passed": passed,
"data": {"coverage_percent": total},
"message": f"Coverage {total}% {'meets' if passed else 'below'} threshold"
}
input_schema:
type: object
required:
- coverage_json
properties:
coverage_json:
type: string
description: JSON string of the coverage report
threshold:
type: number
description: Minimum coverage percentage required
default: 80
resource_slots:
- name: project_repo
resource_type: git-checkout
access: read_only
description: Repository under test
binding: contextual
timeout: 120
@@ -0,0 +1,29 @@
# Example: Wrapped validation
# Wraps an existing tool and transforms its output into a pass/fail result.
name: qa/lint-check
description: Validate that linting passes by wrapping the lint tool
wraps: devops/run-linter
mode: required
transform: |
def transform(result):
errors = result.get("errors", [])
passed = len(errors) == 0
return {
"passed": passed,
"data": {"error_count": len(errors), "errors": errors[:5]},
"message": f"{len(errors)} lint errors found" if errors else "No lint errors"
}
argument_mapping:
strict: true
config_path: .eslintrc.json
resource_slots:
- name: source_code
resource_type: git-checkout
access: read_only
description: Source code to lint
binding: contextual
timeout: 180
+927
View File
@@ -0,0 +1,927 @@
"""Step definitions for Tool and Validation domain model tests."""
from typing import Any
from behave import then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.domain.models.core.tool import (
BindingMode,
CheckpointScope,
ResourceAccessMode,
ResourceSlot,
Tool,
ToolCapability,
ToolLifecycle,
ToolSource,
ToolType,
Validation,
ValidationMode,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_tool(**overrides: Any) -> Tool:
"""Create a Tool with sensible defaults, allowing overrides."""
defaults: dict[str, Any] = {
"name": "local/test-tool",
"description": "A test tool",
"source": ToolSource.BUILTIN,
}
defaults.update(overrides)
return Tool(**defaults)
def _tool_config(**overrides: Any) -> dict[str, Any]:
"""Return a minimal valid config dict for Tool.from_config."""
cfg: dict[str, Any] = {
"name": "local/my-tool",
"description": "A tool from config",
"source": "builtin",
}
cfg.update(overrides)
return cfg
def _validation_config(**overrides: Any) -> dict[str, Any]:
"""Return a minimal valid config dict for Validation.from_config."""
cfg: dict[str, Any] = {
"name": "qa/my-validation",
"description": "A validation from config",
"source": "builtin",
}
cfg.update(overrides)
return cfg
# ---------------------------------------------------------------------------
# Tool Creation Steps
# ---------------------------------------------------------------------------
@when('I create a tool with name "{name}" source "{source}" and code "{code}"')
def step_create_tool_custom(
context: Context, name: str, source: str, code: str
) -> None:
"""Create a custom tool with inline code."""
context.tool_model = Tool(
name=name,
description="A custom test tool",
source=ToolSource(source),
code=code,
)
context.tool_model_error = None
@when('I create a tool with name "{name}" and source "{source}"')
def step_create_tool_builtin(context: Context, name: str, source: str) -> None:
"""Create a tool with a given source."""
context.tool_model = _make_tool(name=name, source=ToolSource(source))
context.tool_model_error = None
@when(
'I create an MCP tool with name "{name}" server "{server}" '
'and tool name "{tool_name}"'
)
def step_create_mcp_tool(
context: Context, name: str, server: str, tool_name: str
) -> None:
"""Create an MCP tool with required fields."""
context.tool_model = Tool(
name=name,
description="An MCP tool",
source=ToolSource.MCP,
mcp_server=server,
mcp_tool_name=tool_name,
)
context.tool_model_error = None
@when('I create an agent skill tool with name "{name}" and path "{path}"')
def step_create_agent_skill_tool(context: Context, name: str, path: str) -> None:
"""Create an agent_skill tool."""
context.tool_model = Tool(
name=name,
description="An agent skill tool",
source=ToolSource.AGENT_SKILL,
agent_skill_path=path,
)
context.tool_model_error = None
# ---------------------------------------------------------------------------
# Tool assertions
# ---------------------------------------------------------------------------
@then("the tool model should be created")
def step_check_tool_created(context: Context) -> None:
"""Verify the tool was created."""
assert context.tool_model is not None, "Tool should be created"
@then('the tool model name should be "{expected}"')
def step_check_tool_name(context: Context, expected: str) -> None:
"""Check the tool name."""
assert context.tool_model.name == expected, (
f"Expected name '{expected}', got '{context.tool_model.name}'"
)
@then('the tool model source should be "{expected}"')
def step_check_tool_source(context: Context, expected: str) -> None:
"""Check the tool source."""
assert context.tool_model.source.value == expected, (
f"Expected source '{expected}', got '{context.tool_model.source.value}'"
)
@then('the tool model type should be "{expected}"')
def step_check_tool_type(context: Context, expected: str) -> None:
"""Check the tool type."""
assert context.tool_model.tool_type.value == expected, (
f"Expected type '{expected}', got '{context.tool_model.tool_type.value}'"
)
@then('the tool model namespace should be "{expected}"')
def step_check_tool_namespace(context: Context, expected: str) -> None:
"""Check the tool namespace property."""
assert context.tool_model.namespace == expected, (
f"Expected namespace '{expected}', got '{context.tool_model.namespace}'"
)
@then('the tool model short name should be "{expected}"')
def step_check_tool_short_name(context: Context, expected: str) -> None:
"""Check the tool short_name property."""
assert context.tool_model.short_name == expected, (
f"Expected short_name '{expected}', got '{context.tool_model.short_name}'"
)
# ---------------------------------------------------------------------------
# Name validation
# ---------------------------------------------------------------------------
@when('I try to create a tool with invalid name "{name}"')
def step_try_create_tool_invalid_name(context: Context, name: str) -> None:
"""Attempt to create a tool with an invalid name."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = _make_tool(name=name)
except ValidationError as e:
context.tool_model_error = e
@then("a tool model validation error should be raised")
def step_check_tool_validation_error(context: Context) -> None:
"""Verify that a validation error was raised."""
assert context.tool_model_error is not None, (
"Expected a validation error to be raised"
)
@then('the tool model error should mention "{text}"')
def step_check_tool_error_message(context: Context, text: str) -> None:
"""Check the error message contains expected text."""
error_str = str(context.tool_model_error)
assert text in error_str, f"Expected error to mention '{text}', got: {error_str}"
# ---------------------------------------------------------------------------
# Source-conditional field requirements
# ---------------------------------------------------------------------------
@when('I try to create a tool with name "{name}" source "{source}" without code')
def step_try_create_custom_without_code(
context: Context, name: str, source: str
) -> None:
"""Attempt to create a custom tool without code."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Tool(
name=name,
description="Missing code",
source=ToolSource(source),
)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create an MCP tool without mcp_server")
def step_try_create_mcp_without_server(context: Context) -> None:
"""Attempt to create an MCP tool without mcp_server."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Tool(
name="local/test",
description="Missing server",
source=ToolSource.MCP,
mcp_tool_name="some_tool",
)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create an MCP tool without mcp_tool_name")
def step_try_create_mcp_without_tool_name(context: Context) -> None:
"""Attempt to create an MCP tool without mcp_tool_name."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Tool(
name="local/test",
description="Missing tool name",
source=ToolSource.MCP,
mcp_server="some_server",
)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create an agent skill tool without path")
def step_try_create_agent_skill_without_path(context: Context) -> None:
"""Attempt to create an agent_skill tool without agent_skill_path."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Tool(
name="local/test",
description="Missing path",
source=ToolSource.AGENT_SKILL,
)
except ValidationError as e:
context.tool_model_error = e
# ---------------------------------------------------------------------------
# Capability constraint enforcement
# ---------------------------------------------------------------------------
@when("I try to create a tool capability with read_only true and writes true")
def step_try_capability_readonly_writes(context: Context) -> None:
"""Attempt to create read_only capability with writes=True."""
context.tool_model_error = None
try:
ToolCapability(read_only=True, writes=True)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create a tool capability with read_only true and checkpointable true")
def step_try_capability_readonly_checkpoint(context: Context) -> None:
"""Attempt to create read_only capability with checkpointable=True."""
context.tool_model_error = None
try:
ToolCapability(read_only=True, checkpointable=True)
except ValidationError as e:
context.tool_model_error = e
@when("I create a tool capability with writes true")
def step_create_capability_with_writes(context: Context) -> None:
"""Create a capability with writes enabled."""
context.tool_capability = ToolCapability(writes=True)
@then("the tool capability should have writes true")
def step_check_capability_writes(context: Context) -> None:
"""Check capability writes is True."""
assert context.tool_capability.writes is True
# ---------------------------------------------------------------------------
# ResourceSlot binding validation
# ---------------------------------------------------------------------------
@when("I try to create a resource slot with static binding and no static resource")
def step_try_slot_static_no_resource(context: Context) -> None:
"""Attempt static binding without static_resource."""
context.tool_model_error = None
try:
ResourceSlot(
name="repo",
resource_type="git-checkout",
access=ResourceAccessMode.READ_ONLY,
binding=BindingMode.STATIC,
)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create a resource slot with contextual binding and static resource set")
def step_try_slot_contextual_with_resource(context: Context) -> None:
"""Attempt contextual binding with static_resource."""
context.tool_model_error = None
try:
ResourceSlot(
name="repo",
resource_type="git-checkout",
access=ResourceAccessMode.READ_ONLY,
binding=BindingMode.CONTEXTUAL,
static_resource="my-repo",
)
except ValidationError as e:
context.tool_model_error = e
@when('I create a resource slot with static binding and static resource "{resource}"')
def step_create_slot_static(context: Context, resource: str) -> None:
"""Create a valid static resource slot."""
context.tool_resource_slot = ResourceSlot(
name="repo",
resource_type="git-checkout",
access=ResourceAccessMode.READ_ONLY,
binding=BindingMode.STATIC,
static_resource=resource,
)
@then("the tool resource slot should be created")
def step_check_slot_created(context: Context) -> None:
"""Verify slot was created."""
assert context.tool_resource_slot is not None
@when('I try to create a resource slot with invalid name "{name}"')
def step_try_slot_invalid_name(context: Context, name: str) -> None:
"""Attempt to create a slot with invalid name."""
context.tool_model_error = None
try:
ResourceSlot(
name=name,
resource_type="git-checkout",
access=ResourceAccessMode.READ_ONLY,
)
except ValidationError as e:
context.tool_model_error = e
# ---------------------------------------------------------------------------
# Validation model creation
# ---------------------------------------------------------------------------
@when('I create a validation with name "{name}" source "{source}" and code "{code}"')
def step_create_validation_custom(
context: Context, name: str, source: str, code: str
) -> None:
"""Create a custom validation."""
context.tool_model = Validation(
name=name,
description="A test validation",
source=ToolSource(source),
code=code,
)
context.tool_model_error = None
@when('I create a validation with name "{name}" source "{source}"')
def step_create_validation_builtin(context: Context, name: str, source: str) -> None:
"""Create a builtin validation."""
context.tool_model = Validation(
name=name,
description="A test validation",
source=ToolSource(source),
)
context.tool_model_error = None
@when('I create an informational validation with name "{name}" source "{source}"')
def step_create_informational_validation(
context: Context, name: str, source: str
) -> None:
"""Create an informational validation."""
context.tool_model = Validation(
name=name,
description="An info validation",
source=ToolSource(source),
mode=ValidationMode.INFORMATIONAL,
)
context.tool_model_error = None
@then("the tool validation capability should be read only")
def step_check_validation_readonly(context: Context) -> None:
"""Check validation capability is read_only."""
assert context.tool_model.capability.read_only is True
@then("the tool validation capability should not have writes")
def step_check_validation_no_writes(context: Context) -> None:
"""Check validation capability has writes=False."""
assert context.tool_model.capability.writes is False
@then("the tool validation capability should not be checkpointable")
def step_check_validation_no_checkpoint(context: Context) -> None:
"""Check validation capability has checkpointable=False."""
assert context.tool_model.capability.checkpointable is False
@then('the tool validation mode should be "{expected}"')
def step_check_validation_mode(context: Context, expected: str) -> None:
"""Check validation mode."""
assert context.tool_model.mode.value == expected, (
f"Expected mode '{expected}', got '{context.tool_model.mode.value}'"
)
# ---------------------------------------------------------------------------
# Validation wraps/transform
# ---------------------------------------------------------------------------
@when("I try to create a wrapped validation without transform")
def step_try_wrapped_validation_no_transform(context: Context) -> None:
"""Attempt wrapped validation without transform."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Validation(
name="qa/bad-wrap",
description="Missing transform",
source=ToolSource.WRAPPED,
wraps="devops/run-linter",
)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create a wrapped validation with code")
def step_try_wrapped_validation_with_code(context: Context) -> None:
"""Attempt wrapped validation with code (mutually exclusive)."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Validation(
name="qa/bad-wrap",
description="Code conflict",
source=ToolSource.WRAPPED,
wraps="devops/run-linter",
transform="def transform(r): return r",
code="print('bad')",
)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create a validation with wraps and wrong source")
def step_try_wrapped_validation_wrong_source(context: Context) -> None:
"""Attempt validation with wraps but source != wrapped."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Validation(
name="qa/bad-wrap",
description="Wrong source",
source=ToolSource.BUILTIN,
wraps="devops/run-linter",
transform="def transform(r): return r",
)
except ValidationError as e:
context.tool_model_error = e
@when("I try to create a validation with argument_mapping but no wraps")
def step_try_validation_argmap_no_wraps(context: Context) -> None:
"""Attempt validation with argument_mapping but no wraps."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Validation(
name="qa/bad-argmap",
description="Argmap without wraps",
source=ToolSource.BUILTIN,
argument_mapping={"key": "value"},
)
except ValidationError as e:
context.tool_model_error = e
@when("I create a valid wrapped validation")
def step_create_valid_wrapped_validation(context: Context) -> None:
"""Create a valid wrapped validation."""
context.tool_model = Validation(
name="qa/lint-check",
description="Lint validation",
source=ToolSource.WRAPPED,
wraps="devops/run-linter",
transform="def transform(r): return {'passed': True}",
argument_mapping={"strict": True},
)
context.tool_model_error = None
@then('the tool validation wraps should be "{expected}"')
def step_check_validation_wraps(context: Context, expected: str) -> None:
"""Check validation wraps field."""
assert context.tool_model.wraps == expected, (
f"Expected wraps '{expected}', got '{context.tool_model.wraps}'"
)
# ---------------------------------------------------------------------------
# from_config loading
# ---------------------------------------------------------------------------
@when('I load a tool from config with name "{name}" source "{source}"')
def step_load_tool_from_config(context: Context, name: str, source: str) -> None:
"""Load a tool from a config dict."""
config = _tool_config(name=name, source=source)
context.tool_model = Tool.from_config(config)
context.tool_model_error = None
@when('I try to load a tool from config missing "{field}"')
def step_try_load_tool_config_missing(context: Context, field: str) -> None:
"""Try loading a tool config with missing field."""
config = _tool_config()
del config[field]
context.tool_model_config_error = None
try:
Tool.from_config(config)
except ValueError as e:
context.tool_model_config_error = e
@then('a tool model config error should be raised with "{field}"')
def step_check_tool_config_error(context: Context, field: str) -> None:
"""Check that a config error mentioning field was raised."""
assert context.tool_model_config_error is not None, (
f"Expected ValueError for missing '{field}'"
)
assert field in str(context.tool_model_config_error), (
f"Expected error to mention '{field}', got: {context.tool_model_config_error}"
)
@when('I load a validation from config with wraps "{wraps}"')
def step_load_validation_from_config_wraps(context: Context, wraps: str) -> None:
"""Load a validation from config with wraps."""
config = _validation_config(
wraps=wraps,
transform="def transform(r): return r",
)
# Remove source since wraps implies wrapped
config.pop("source", None)
context.tool_model = Validation.from_config(config)
context.tool_model_error = None
@when('I try to load a validation config missing "{field}"')
def step_try_load_validation_config_missing(context: Context, field: str) -> None:
"""Try loading a validation config with missing field."""
config = _validation_config()
del config[field]
context.tool_model_config_error = None
try:
Validation.from_config(config)
except ValueError as e:
context.tool_model_config_error = e
# ---------------------------------------------------------------------------
# as_cli_dict
# ---------------------------------------------------------------------------
@when("I create a tool and call as_cli_dict")
def step_create_tool_cli_dict(context: Context) -> None:
"""Create a tool and get CLI dict."""
context.tool_model = _make_tool()
context.tool_cli_dict = context.tool_model.as_cli_dict()
@when("I create a validation and call as_cli_dict")
def step_create_validation_cli_dict(context: Context) -> None:
"""Create a validation and get CLI dict."""
context.tool_model = Validation(
name="qa/test-check",
description="A validation",
source=ToolSource.BUILTIN,
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
@then('the tool cli dict should have key "{key}"')
def step_check_tool_cli_dict_key(context: Context, key: str) -> None:
"""Check CLI dict has expected key."""
assert key in context.tool_cli_dict, (
f"Expected key '{key}' in cli dict, keys: {list(context.tool_cli_dict)}"
)
# ---------------------------------------------------------------------------
# Timeout
# ---------------------------------------------------------------------------
@when("I try to create a tool with timeout {timeout:d}")
def step_try_create_tool_bad_timeout(context: Context, timeout: int) -> None:
"""Attempt to create a tool with invalid timeout."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = _make_tool(timeout=timeout)
except ValidationError as e:
context.tool_model_error = e
@then("the tool model timeout should be {expected:d}")
def step_check_tool_timeout(context: Context, expected: int) -> None:
"""Check tool timeout."""
assert context.tool_model.timeout == expected, (
f"Expected timeout {expected}, got {context.tool_model.timeout}"
)
# ---------------------------------------------------------------------------
# ToolLifecycle
# ---------------------------------------------------------------------------
@when('I create a tool lifecycle with discover "{discover}"')
def step_create_lifecycle(context: Context, discover: str) -> None:
"""Create a ToolLifecycle."""
context.tool_lifecycle = ToolLifecycle(discover=discover)
@then('the tool lifecycle discover should be "{expected}"')
def step_check_lifecycle_discover(context: Context, expected: str) -> None:
"""Check lifecycle discover hook."""
assert context.tool_lifecycle.discover == expected
# ---------------------------------------------------------------------------
# Enum checks
# ---------------------------------------------------------------------------
@then('the ToolSource enum should have values "{values}"')
def step_check_tool_source_enum(context: Context, values: str) -> None:
"""Verify ToolSource enum values."""
expected = set(values.split(","))
actual = {e.value for e in ToolSource}
assert actual == expected, f"Expected {expected}, got {actual}"
@then('the ToolType enum should have values "{values}"')
def step_check_tool_type_enum(context: Context, values: str) -> None:
"""Verify ToolType enum values."""
expected = set(values.split(","))
actual = {e.value for e in ToolType}
assert actual == expected, f"Expected {expected}, got {actual}"
@then('the ValidationMode enum should have values "{values}"')
def step_check_validation_mode_enum(context: Context, values: str) -> None:
"""Verify ValidationMode enum values."""
expected = set(values.split(","))
actual = {e.value for e in ValidationMode}
assert actual == expected, f"Expected {expected}, got {actual}"
@then('the CheckpointScope enum should have values "{values}"')
def step_check_checkpoint_scope_enum(context: Context, values: str) -> None:
"""Verify CheckpointScope enum values."""
expected = set(values.split(","))
actual = {e.value for e in CheckpointScope}
assert actual == expected, f"Expected {expected}, got {actual}"
@then('the BindingMode enum should have values "{values}"')
def step_check_binding_mode_enum(context: Context, values: str) -> None:
"""Verify BindingMode enum values."""
expected = set(values.split(","))
actual = {e.value for e in BindingMode}
assert actual == expected, f"Expected {expected}, got {actual}"
@then('the ResourceAccessMode enum should have values "{values}"')
def step_check_access_mode_enum(context: Context, values: str) -> None:
"""Verify ResourceAccessMode enum values."""
expected = set(values.split(","))
actual = {e.value for e in ResourceAccessMode}
assert actual == expected, f"Expected {expected}, got {actual}"
# ---------------------------------------------------------------------------
# from_config with resource slots
# ---------------------------------------------------------------------------
@when("I load a tool from config with resource slots")
def step_load_tool_config_with_slots(context: Context) -> None:
"""Load a tool from config that includes resource slots."""
config = _tool_config(
resource_slots=[
{
"name": "repo",
"resource_type": "git-checkout",
"access": "read_only",
},
],
)
context.tool_model = Tool.from_config(config)
context.tool_model_error = None
@then("the tool model should have {count:d} resource slot")
def step_check_tool_slot_count(context: Context, count: int) -> None:
"""Check tool resource slot count."""
actual = len(context.tool_model.resource_slots)
assert actual == count, f"Expected {count} slots, got {actual}"
# ---------------------------------------------------------------------------
# Empty description
# ---------------------------------------------------------------------------
@when("I try to create a tool with empty description")
def step_try_create_tool_empty_desc(context: Context) -> None:
"""Attempt to create a tool with empty description."""
context.tool_model_error = None
context.tool_model = None
try:
context.tool_model = Tool(
name="local/bad-desc",
description="",
source=ToolSource.BUILTIN,
)
except ValidationError as e:
context.tool_model_error = e
# ---------------------------------------------------------------------------
# as_cli_dict coverage for optional fields
# ---------------------------------------------------------------------------
@when("I create a custom tool and call as_cli_dict")
def step_create_custom_tool_cli_dict(context: Context) -> None:
"""Create a custom tool and get CLI dict (covers code branch)."""
context.tool_model = Tool(
name="local/custom-cli",
description="Custom tool for CLI dict",
source=ToolSource.CUSTOM,
code="return True",
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
@when("I create an MCP tool and call as_cli_dict")
def step_create_mcp_tool_cli_dict(context: Context) -> None:
"""Create an MCP tool and get CLI dict (covers mcp fields)."""
context.tool_model = Tool(
name="devops/mcp-cli",
description="MCP tool for CLI dict",
source=ToolSource.MCP,
mcp_server="devops-mcp",
mcp_tool_name="docker_build",
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
@when("I create an agent skill tool and call as_cli_dict")
def step_create_agent_skill_tool_cli_dict(context: Context) -> None:
"""Create an agent skill tool and get CLI dict."""
context.tool_model = Tool(
name="skills/fmt-cli",
description="Agent skill for CLI dict",
source=ToolSource.AGENT_SKILL,
agent_skill_path="/skills/fmt",
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
@when("I create a tool with schemas and call as_cli_dict")
def step_create_tool_with_schemas_cli_dict(context: Context) -> None:
"""Create a tool with schemas and get CLI dict."""
context.tool_model = Tool(
name="local/schema-cli",
description="Tool with schemas",
source=ToolSource.BUILTIN,
input_schema={"type": "object"},
output_schema={"type": "object"},
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
@when("I create a tool with side effects and cost profile and call as_cli_dict")
def step_create_tool_with_effects_cli_dict(context: Context) -> None:
"""Create a tool with side effects and cost profile."""
context.tool_model = Tool(
name="local/effects-cli",
description="Tool with effects",
source=ToolSource.BUILTIN,
capability=ToolCapability(
side_effects=["network", "filesystem"],
cost_profile="high",
),
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
@then('the tool cli dict capability should have key "{key}"')
def step_check_tool_cli_dict_cap_key(context: Context, key: str) -> None:
"""Check CLI dict capability sub-dict has expected key."""
cap = context.tool_cli_dict["capability"]
assert key in cap, f"Expected key '{key}' in capability dict, keys: {list(cap)}"
@when("I create a tool with resource slots and call as_cli_dict")
def step_create_tool_with_slots_cli_dict(context: Context) -> None:
"""Create a tool with resource slots and get CLI dict."""
context.tool_model = Tool(
name="local/slots-cli",
description="Tool with slots",
source=ToolSource.BUILTIN,
resource_slots=[
ResourceSlot(
name="repo",
resource_type="git-checkout",
access=ResourceAccessMode.READ_ONLY,
),
],
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
# ---------------------------------------------------------------------------
# Validation as_cli_dict wraps fields
# ---------------------------------------------------------------------------
@when("I create a wrapped validation and call as_cli_dict")
def step_create_wrapped_validation_cli_dict(context: Context) -> None:
"""Create a wrapped validation and get CLI dict."""
context.tool_model = Validation(
name="qa/wrap-cli",
description="Wrapped validation for CLI dict",
source=ToolSource.WRAPPED,
wraps="devops/run-linter",
transform="def transform(r): return {'passed': True}",
argument_mapping={"strict": True},
)
context.tool_cli_dict = context.tool_model.as_cli_dict()
# ---------------------------------------------------------------------------
# Validation from_config without wraps
# ---------------------------------------------------------------------------
@when("I try to load a validation config without source or wraps")
def step_try_load_validation_no_source_no_wraps(context: Context) -> None:
"""Try loading a validation config with neither source nor wraps."""
config = _validation_config()
del config["source"]
context.tool_model_config_error = None
try:
Validation.from_config(config)
except ValueError as e:
context.tool_model_config_error = e
@when('I load a validation from config with source "{source}"')
def step_load_validation_from_config_source(context: Context, source: str) -> None:
"""Load a validation from config with explicit source."""
config = _validation_config(source=source)
context.tool_model = Validation.from_config(config)
context.tool_model_error = None
# ---------------------------------------------------------------------------
# Validation forces read_only on writable capability
# ---------------------------------------------------------------------------
@when("I create a validation with writes capability")
def step_create_validation_with_writes(context: Context) -> None:
"""Create a validation that has writes in capability (will be forced)."""
context.tool_model = Validation(
name="qa/writable-val",
description="Validation with writes cap",
source=ToolSource.BUILTIN,
capability=ToolCapability(
writes=True,
checkpointable=True,
),
)
context.tool_model_error = None
+327
View File
@@ -0,0 +1,327 @@
Feature: Tool and Validation Domain Models
As a developer
I want tool and validation domain models for the tool registry
So that tools can be registered, validated, and managed via CLI
# ---- Tool Creation ----
Scenario: Create a custom tool with valid fields
When I create a tool with name "local/line-counter" source "custom" and code "print('hello')"
Then the tool model should be created
And the tool model name should be "local/line-counter"
And the tool model source should be "custom"
And the tool model type should be "tool"
And the tool model namespace should be "local"
And the tool model short name should be "line-counter"
Scenario: Create a builtin tool
When I create a tool with name "builtin/file-read" and source "builtin"
Then the tool model should be created
And the tool model source should be "builtin"
Scenario: Create an MCP tool with required fields
When I create an MCP tool with name "devops/docker-build" server "devops-mcp" and tool name "docker_build"
Then the tool model should be created
And the tool model source should be "mcp"
Scenario: Create an agent_skill tool
When I create an agent skill tool with name "skills/formatter" and path "/skills/formatter"
Then the tool model should be created
And the tool model source should be "agent_skill"
# ---- Tool Name Validation ----
Scenario: Tool name must match namespace/name pattern
When I try to create a tool with invalid name "no-namespace"
Then a tool model validation error should be raised
And the tool model error should mention "namespace/name"
Scenario: Tool name rejects spaces
When I try to create a tool with invalid name "local/bad name"
Then a tool model validation error should be raised
Scenario: Tool name rejects double slashes
When I try to create a tool with invalid name "local//bad"
Then a tool model validation error should be raised
Scenario: Tool name allows hyphens and underscores
When I create a tool with name "my-ns/my_tool" and source "builtin"
Then the tool model should be created
And the tool model namespace should be "my-ns"
And the tool model short name should be "my_tool"
# ---- Source-conditional field requirements ----
Scenario: Custom source requires code
When I try to create a tool with name "local/test" source "custom" without code
Then a tool model validation error should be raised
And the tool model error should mention "code is required"
Scenario: MCP source requires mcp_server
When I try to create an MCP tool without mcp_server
Then a tool model validation error should be raised
And the tool model error should mention "mcp_server is required"
Scenario: MCP source requires mcp_tool_name
When I try to create an MCP tool without mcp_tool_name
Then a tool model validation error should be raised
And the tool model error should mention "mcp_tool_name is required"
Scenario: Agent skill source requires agent_skill_path
When I try to create an agent skill tool without path
Then a tool model validation error should be raised
And the tool model error should mention "agent_skill_path is required"
# ---- Capability constraint enforcement ----
Scenario: Read-only capability enforces writes=false
When I try to create a tool capability with read_only true and writes true
Then a tool model validation error should be raised
And the tool model error should mention "read_only tool cannot have writes"
Scenario: Read-only capability enforces checkpointable=false
When I try to create a tool capability with read_only true and checkpointable true
Then a tool model validation error should be raised
And the tool model error should mention "read_only tool cannot have checkpointable"
Scenario: Non-read-only capability allows writes
When I create a tool capability with writes true
Then the tool capability should have writes true
# ---- ResourceSlot binding validation ----
Scenario: Static binding requires static_resource
When I try to create a resource slot with static binding and no static resource
Then a tool model validation error should be raised
And the tool model error should mention "static_resource is required"
Scenario: Non-static binding forbids static_resource
When I try to create a resource slot with contextual binding and static resource set
Then a tool model validation error should be raised
And the tool model error should mention "static_resource must not be set"
Scenario: Static binding with static_resource succeeds
When I create a resource slot with static binding and static resource "my-repo"
Then the tool resource slot should be created
Scenario: ResourceSlot name must be valid identifier
When I try to create a resource slot with invalid name "123bad"
Then a tool model validation error should be raised
And the tool model error should mention "valid Python identifier"
# ---- Validation model creation ----
Scenario: Create a custom validation with forced read-only
When I create a validation with name "qa/coverage-check" source "custom" and code "return True"
Then the tool model should be created
And the tool model type should be "validation"
And the tool validation capability should be read only
And the tool validation capability should not have writes
And the tool validation capability should not be checkpointable
Scenario: Validation forces tool_type to validation
When I create a validation with name "qa/test-check" source "builtin"
Then the tool model type should be "validation"
Scenario: Validation mode defaults to required
When I create a validation with name "qa/check" source "builtin"
Then the tool validation mode should be "required"
Scenario: Validation can be informational
When I create an informational validation with name "qa/info-check" source "builtin"
Then the tool validation mode should be "informational"
# ---- Validation wraps/transform requirements ----
Scenario: Wrapped validation requires transform
When I try to create a wrapped validation without transform
Then a tool model validation error should be raised
And the tool model error should mention "transform is required"
Scenario: Wrapped validation forbids code
When I try to create a wrapped validation with code
Then a tool model validation error should be raised
And the tool model error should mention "code must not be set"
Scenario: Wrapped validation requires source=wrapped
When I try to create a validation with wraps and wrong source
Then a tool model validation error should be raised
And the tool model error should mention "source must be 'wrapped'"
Scenario: argument_mapping only valid with wraps
When I try to create a validation with argument_mapping but no wraps
Then a tool model validation error should be raised
And the tool model error should mention "argument_mapping is only valid"
Scenario: Valid wrapped validation
When I create a valid wrapped validation
Then the tool model should be created
And the tool model type should be "validation"
And the tool validation wraps should be "devops/run-linter"
# ---- from_config loading ----
Scenario: Load tool from config dict
When I load a tool from config with name "local/my-tool" source "builtin"
Then the tool model should be created
And the tool model name should be "local/my-tool"
Scenario: Load tool from config missing name raises error
When I try to load a tool from config missing "name"
Then a tool model config error should be raised with "name"
Scenario: Load tool from config missing description raises error
When I try to load a tool from config missing "description"
Then a tool model config error should be raised with "description"
Scenario: Load tool from config missing source raises error
When I try to load a tool from config missing "source"
Then a tool model config error should be raised with "source"
Scenario: Load validation from config with wraps
When I load a validation from config with wraps "devops/linter"
Then the tool model should be created
And the tool model source should be "wrapped"
And the tool validation wraps should be "devops/linter"
Scenario: Load validation from config missing name raises error
When I try to load a validation config missing "name"
Then a tool model config error should be raised with "name"
# ---- as_cli_dict output stability ----
Scenario: Tool as_cli_dict has required keys
When I create a tool and call as_cli_dict
Then the tool cli dict should have key "name"
And the tool cli dict should have key "description"
And the tool cli dict should have key "source"
And the tool cli dict should have key "tool_type"
And the tool cli dict should have key "namespace"
And the tool cli dict should have key "short_name"
And the tool cli dict should have key "capability"
And the tool cli dict should have key "timeout"
Scenario: Validation as_cli_dict includes mode
When I create a validation and call as_cli_dict
Then the tool cli dict should have key "mode"
# ---- Tool timeout validation ----
Scenario: Tool timeout must be >= 1
When I try to create a tool with timeout 0
Then a tool model validation error should be raised
Scenario: Tool timeout default is 300
When I create a tool with name "local/default-timeout" and source "builtin"
Then the tool model timeout should be 300
# ---- ToolLifecycle ----
Scenario: ToolLifecycle accepts optional hooks
When I create a tool lifecycle with discover "my.discover"
Then the tool lifecycle discover should be "my.discover"
# ---- Enum values ----
Scenario: ToolSource enum has all expected values
Then the ToolSource enum should have values "mcp,agent_skill,builtin,custom,wrapped"
Scenario: ToolType enum has all expected values
Then the ToolType enum should have values "tool,validation"
Scenario: ValidationMode enum has all expected values
Then the ValidationMode enum should have values "required,informational"
Scenario: CheckpointScope enum has all expected values
Then the CheckpointScope enum should have values "file,transaction,snapshot,composite"
Scenario: BindingMode enum has all expected values
Then the BindingMode enum should have values "contextual,static,parameter"
Scenario: ResourceAccessMode enum has all expected values
Then the ResourceAccessMode enum should have values "read_only,read_write"
# ---- Tool with resource slots via from_config ----
Scenario: Load tool from config with resource slots
When I load a tool from config with resource slots
Then the tool model should be created
And the tool model should have 1 resource slot
# ---- Validation from_config missing description ----
Scenario: Load validation from config missing description raises error
When I try to load a validation config missing "description"
Then a tool model config error should be raised with "description"
# ---- Tool description validation ----
Scenario: Tool description must not be empty
When I try to create a tool with empty description
Then a tool model validation error should be raised
# ---- Invalid scenarios ----
Scenario: Tool with conflicting capability flags
When I try to create a tool capability with read_only true and writes true
Then a tool model validation error should be raised
Scenario: Validation wraps with custom source is rejected
When I try to create a validation with wraps and wrong source
Then a tool model validation error should be raised
# ---- as_cli_dict coverage for optional fields ----
Scenario: Tool as_cli_dict includes code when set
When I create a custom tool and call as_cli_dict
Then the tool cli dict should have key "code"
Scenario: Tool as_cli_dict includes mcp fields when set
When I create an MCP tool and call as_cli_dict
Then the tool cli dict should have key "mcp_server"
And the tool cli dict should have key "mcp_tool_name"
Scenario: Tool as_cli_dict includes agent_skill_path when set
When I create an agent skill tool and call as_cli_dict
Then the tool cli dict should have key "agent_skill_path"
Scenario: Tool as_cli_dict includes schemas when set
When I create a tool with schemas and call as_cli_dict
Then the tool cli dict should have key "input_schema"
And the tool cli dict should have key "output_schema"
Scenario: Tool as_cli_dict includes side_effects and cost_profile
When I create a tool with side effects and cost profile and call as_cli_dict
Then the tool cli dict capability should have key "side_effects"
And the tool cli dict capability should have key "cost_profile"
Scenario: Tool as_cli_dict includes resource slots when set
When I create a tool with resource slots and call as_cli_dict
Then the tool cli dict should have key "resource_slots"
# ---- Validation as_cli_dict wraps fields ----
Scenario: Validation as_cli_dict includes wraps and transform
When I create a wrapped validation and call as_cli_dict
Then the tool cli dict should have key "wraps"
And the tool cli dict should have key "transform"
And the tool cli dict should have key "argument_mapping"
# ---- Validation from_config without wraps ----
Scenario: Load validation from config without wraps needs source
When I try to load a validation config without source or wraps
Then a tool model config error should be raised with "source"
Scenario: Load validation from config with explicit source
When I load a validation from config with source "builtin"
Then the tool model should be created
And the tool model source should be "builtin"
# ---- Validation forces read_only on writable capability ----
Scenario: Validation forces read_only on writable capability
When I create a validation with writes capability
Then the tool validation capability should be read only
And the tool validation capability should not have writes
And the tool validation capability should not be checkpointable
@@ -69,11 +69,28 @@ from cleveragents.domain.models.core.resource import (
SandboxStrategy,
)
# Tool and Validation domain models
from cleveragents.domain.models.core.tool import (
BindingMode,
CheckpointScope,
ResourceAccessMode,
ResourceSlot,
Tool,
ToolCapability,
ToolLifecycle,
ToolSource,
ToolType,
Validation,
ValidationMode,
)
__all__ = [
"ActionState",
"Actor",
"BindingMode",
"Change",
"ChangeSet",
"CheckpointScope",
"CloudBillingFields",
"Context",
"ContextConfig",
@@ -112,11 +129,20 @@ __all__ = [
"ProjectSettings",
"ProjectStats",
"Resource",
"ResourceAccessMode",
"ResourceCapabilities",
"ResourceSlot",
"SandboxStrategy",
"SummaryForUpdateContextParams",
"TemporalScope",
"Tool",
"ToolCapability",
"ToolLifecycle",
"ToolSource",
"ToolType",
"User",
"Validation",
"ValidationMode",
"can_transition",
"parse_namespaced_name",
]
+623
View File
@@ -0,0 +1,623 @@
"""Tool and Validation domain models for CleverAgents v3.
A Tool is the atomic unit of execution: a namespaced, independently registered
callable operation. Defined by JSON Schema inputs/outputs, capability metadata
(``read_only``, ``writes``, ``checkpointable``), and typed resource slots.
A Validation extends Tool with pass/fail semantics: always read-only, always
``writes=False`` and ``checkpointable=False``. May wrap an existing tool via
``wraps`` + ``transform`` to reuse its implementation.
Tools and Validations share the same naming namespace -- no collisions allowed.
Based on docs/specification.md and ADR-004 (Pydantic Validation).
"""
from __future__ import annotations
import re
from collections import OrderedDict
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
# ---------------------------------------------------------------------------
# Regex patterns
# ---------------------------------------------------------------------------
_TOOL_NAME_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+/[a-zA-Z0-9_-]+$")
# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------
class ToolSource(StrEnum):
"""Source of a tool implementation.
Determines which fields are required on the Tool model:
- ``mcp``: requires ``mcp_server`` and ``mcp_tool_name``
- ``agent_skill``: requires ``agent_skill_path``
- ``builtin``: provided by the runtime
- ``custom``: requires inline ``code``
- ``wrapped``: validation-only; set implicitly when a Validation wraps
an existing tool
"""
MCP = "mcp"
AGENT_SKILL = "agent_skill"
BUILTIN = "builtin"
CUSTOM = "custom"
WRAPPED = "wrapped"
class ToolType(StrEnum):
"""Discriminator for the tool registry.
Tools and Validations share the same namespace but are distinguished by
this field for filtering and listing.
"""
TOOL = "tool"
VALIDATION = "validation"
class ValidationMode(StrEnum):
"""Controls whether validation failure blocks execution.
- ``required``: failure blocks the operation
- ``informational``: failure is reported but does not block
"""
REQUIRED = "required"
INFORMATIONAL = "informational"
class ResourceAccessMode(StrEnum):
"""Access level a tool has on a bound resource."""
READ_ONLY = "read_only"
READ_WRITE = "read_write"
class BindingMode(StrEnum):
"""How a resource slot is resolved to a concrete resource.
- ``contextual``: resolved from the plan's project at invocation time
- ``static``: hardcoded at registration via ``static_resource``
- ``parameter``: passed as a parameter at invocation time
"""
CONTEXTUAL = "contextual"
STATIC = "static"
PARAMETER = "parameter"
class CheckpointScope(StrEnum):
"""Granularity of checkpoint support for a tool."""
FILE = "file"
TRANSACTION = "transaction"
SNAPSHOT = "snapshot"
COMPOSITE = "composite"
# ---------------------------------------------------------------------------
# ResourceSlot
# ---------------------------------------------------------------------------
class ResourceSlot(BaseModel):
"""Declaration of a resource binding for a tool.
Each slot names a typed resource the tool operates on. The binding mode
determines how the concrete resource is resolved at invocation time.
"""
name: str = Field(
..., min_length=1, description="Slot name (valid Python identifier)"
)
resource_type: str = Field(
..., min_length=1, description="Resource type name (e.g. 'git-checkout')"
)
access: ResourceAccessMode = Field(..., description="Access level on the resource")
description: str = Field("", description="Human-readable description of the slot")
binding: BindingMode = Field(
BindingMode.CONTEXTUAL, description="How the resource is resolved"
)
static_resource: str | None = Field(
default=None, description="Resource name when binding=static"
)
required: bool = Field(True, description="Whether the resource must be present")
@field_validator("name")
@classmethod
def validate_name(cls: type[ResourceSlot], v: str) -> str:
"""Validate slot name is a valid Python identifier."""
if not v.isidentifier():
raise ValueError(
"ResourceSlot name must be a valid Python identifier "
"(alphanumeric and underscores, not starting with a digit)"
)
return v
@model_validator(mode="after")
def _validate_static_binding(self) -> ResourceSlot:
"""Enforce static_resource rules based on binding mode."""
if self.binding == BindingMode.STATIC and not self.static_resource:
raise ValueError("static_resource is required when binding is 'static'")
if self.binding != BindingMode.STATIC and self.static_resource is not None:
raise ValueError(
"static_resource must not be set when binding is not 'static'"
)
return self
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# ToolCapability
# ---------------------------------------------------------------------------
class ToolCapability(BaseModel):
"""Capability metadata describing what a tool can and cannot do.
Used by the scheduler, sandbox, and safety layer to make decisions
about checkpoint management, human approval, and cost tracking.
"""
read_only: bool = Field(default=False, description="Tool only reads, never writes")
writes: bool = Field(default=False, description="Tool can write to resources")
write_scope: str | None = Field(
default=None, description="Scope of writes (e.g. 'file', 'database')"
)
checkpointable: bool = Field(
default=False,
description="Tool supports checkpoint/rollback",
)
checkpoint_scope: CheckpointScope | None = Field(
default=None, description="Granularity of checkpoint support"
)
side_effects: list[str] = Field(
default_factory=list,
description="Known side effects (e.g. 'network', 'filesystem')",
)
idempotent: bool = Field(
default=False,
description="Safe to re-run without side effects",
)
unsafe: bool = Field(default=False, description="Requires extra safety checks")
human_approval_required: bool = Field(
default=False,
description="Requires explicit human approval before execution",
)
cost_profile: str | None = Field(
default=None, description="Named cost profile for budgeting"
)
@model_validator(mode="after")
def _enforce_read_only(self) -> ToolCapability:
"""If read_only is True, writes and checkpointable must be False."""
if self.read_only:
if self.writes:
raise ValueError("read_only tool cannot have writes=True")
if self.checkpointable:
raise ValueError("read_only tool cannot have checkpointable=True")
return self
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# ToolLifecycle
# ---------------------------------------------------------------------------
class ToolLifecycle(BaseModel):
"""Optional lifecycle hooks for a tool.
Each hook is a dotted Python path or shell command string that the
runtime invokes at the corresponding lifecycle stage.
"""
discover: str | None = Field(default=None, description="Hook for tool discovery")
activate: str | None = Field(default=None, description="Hook for tool activation")
deactivate: str | None = Field(
default=None, description="Hook for tool deactivation"
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# Tool
# ---------------------------------------------------------------------------
class Tool(BaseModel):
"""Domain model for a Tool.
A Tool is the atomic unit of execution in CleverAgents -- a namespaced,
independently registered callable operation. Created via
``agents tool add --config <file>``.
The **namespaced name** (``namespace/short_name``) is the unique identifier.
Tools and Validations share the same namespace; no collisions are allowed.
"""
# Identity -- namespace/name format
name: str = Field(
...,
min_length=1,
description="Namespaced tool name (namespace/short_name)",
)
description: str = Field(
...,
min_length=1,
description="Short description of what the tool does",
)
# Source
source: ToolSource = Field(
..., description="Where the tool implementation comes from"
)
tool_type: ToolType = Field(
ToolType.TOOL, description="Discriminator: tool or validation"
)
# Source-specific fields
code: str | None = Field(
default=None, description="Inline Python code (required for source=custom)"
)
mcp_server: str | None = Field(
default=None, description="MCP server name (required for source=mcp)"
)
mcp_tool_name: str | None = Field(
default=None,
description="Tool name on the MCP server (required for source=mcp)",
)
agent_skill_path: str | None = Field(
default=None,
description=(
"Path to Agent Skills Standard folder (required for source=agent_skill)"
),
)
# Schemas
input_schema: dict[str, Any] | None = Field(
default=None, description="JSON Schema for tool inputs"
)
output_schema: dict[str, Any] | None = Field(
default=None, description="JSON Schema for tool outputs"
)
# Capability & resource bindings
capability: ToolCapability = Field(
default_factory=lambda: ToolCapability(),
description="Capability metadata for the tool",
)
resource_slots: list[ResourceSlot] = Field(
default_factory=list,
description="Resource slot declarations for the tool",
)
# Lifecycle
lifecycle: ToolLifecycle | None = Field(
default=None, description="Optional lifecycle hooks"
)
# Execution
timeout: int = Field(300, ge=1, description="Execution timeout in seconds")
# -- Name format validation ------------------------------------------------
@field_validator("name")
@classmethod
def validate_name_format(cls: type[Tool], v: str) -> str:
"""Validate tool name matches namespace/name pattern."""
if not _TOOL_NAME_PATTERN.match(v):
raise ValueError(
f"Tool name must match 'namespace/name' pattern "
f"(alphanumeric, hyphens, underscores): got '{v}'"
)
return v
# -- Source-conditional field validation ------------------------------------
@model_validator(mode="after")
def _validate_source_fields(self) -> Tool:
"""Enforce source-conditional field requirements."""
if self.source == ToolSource.CUSTOM and not self.code:
raise ValueError("code is required when source is 'custom'")
if self.source == ToolSource.MCP:
if not self.mcp_server:
raise ValueError("mcp_server is required when source is 'mcp'")
if not self.mcp_tool_name:
raise ValueError("mcp_tool_name is required when source is 'mcp'")
if self.source == ToolSource.AGENT_SKILL and not self.agent_skill_path:
raise ValueError(
"agent_skill_path is required when source is 'agent_skill'"
)
return self
# -- Properties ------------------------------------------------------------
@property
def namespace(self) -> str:
"""Extract the namespace portion from the name."""
return self.name.split("/", 1)[0]
@property
def short_name(self) -> str:
"""Extract the short name portion from the name."""
return self.name.split("/", 1)[1]
# -- Factories & serialization --------------------------------------------
@classmethod
def from_config(cls, config: dict[str, Any]) -> Tool:
"""Create a Tool from a YAML configuration dict.
The config follows the spec schema (see ``docs/specification.md``
Tool Configuration Files section).
"""
if "name" not in config:
raise ValueError("Tool config must include 'name'")
if "description" not in config:
raise ValueError("Tool config must include 'description'")
if "source" not in config:
raise ValueError("Tool config must include 'source'")
# Build capability from nested dict
cap_data = config.get("capability")
capability = (
ToolCapability.model_validate(cap_data) if cap_data else ToolCapability()
)
# Build resource slots
slots_data: list[dict[str, Any]] = config.get("resource_slots", [])
resource_slots = [ResourceSlot.model_validate(s) for s in slots_data]
# Build lifecycle
lc_data = config.get("lifecycle")
lifecycle = ToolLifecycle.model_validate(lc_data) if lc_data else None
return cls(
name=config["name"],
description=config["description"],
source=ToolSource(config["source"]),
tool_type=ToolType(config.get("tool_type", "tool")),
code=config.get("code"),
mcp_server=config.get("mcp_server"),
mcp_tool_name=config.get("mcp_tool_name"),
agent_skill_path=config.get("agent_skill_path"),
input_schema=config.get("input_schema"),
output_schema=config.get("output_schema"),
capability=capability,
resource_slots=resource_slots,
lifecycle=lifecycle,
timeout=config.get("timeout", 300),
)
def as_cli_dict(self) -> OrderedDict[str, Any]:
"""Return a stable-ordered dictionary for CLI rendering.
Keys are ordered for consistent display across output formats.
"""
result: OrderedDict[str, Any] = OrderedDict()
result["name"] = self.name
result["description"] = self.description
result["source"] = self.source.value
result["tool_type"] = self.tool_type.value
result["namespace"] = self.namespace
result["short_name"] = self.short_name
if self.code:
result["code"] = self.code
if self.mcp_server:
result["mcp_server"] = self.mcp_server
if self.mcp_tool_name:
result["mcp_tool_name"] = self.mcp_tool_name
if self.agent_skill_path:
result["agent_skill_path"] = self.agent_skill_path
if self.input_schema:
result["input_schema"] = self.input_schema
if self.output_schema:
result["output_schema"] = self.output_schema
cap_dict: dict[str, Any] = {
"read_only": self.capability.read_only,
"writes": self.capability.writes,
"checkpointable": self.capability.checkpointable,
"idempotent": self.capability.idempotent,
"unsafe": self.capability.unsafe,
"human_approval_required": self.capability.human_approval_required,
}
if self.capability.side_effects:
cap_dict["side_effects"] = list(self.capability.side_effects)
if self.capability.cost_profile:
cap_dict["cost_profile"] = self.capability.cost_profile
result["capability"] = cap_dict
if self.resource_slots:
result["resource_slots"] = [
{
"name": slot.name,
"resource_type": slot.resource_type,
"access": slot.access.value,
"binding": slot.binding.value,
"required": slot.required,
}
for slot in self.resource_slots
]
result["timeout"] = self.timeout
return result
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
use_enum_values=False,
)
# ---------------------------------------------------------------------------
# Validation
# ---------------------------------------------------------------------------
class Validation(Tool):
"""Domain model for a Validation.
A Validation extends Tool with pass/fail semantics: always read-only,
``writes=False``, ``checkpointable=False``. May wrap an existing tool
via ``wraps`` + ``transform`` to reuse its implementation.
Shares the Tool naming namespace -- no collisions allowed.
Managed via ``agents validation add/attach/detach``.
"""
mode: ValidationMode = Field(
ValidationMode.REQUIRED,
description="Whether failure blocks execution",
)
wraps: str | None = Field(
default=None,
description="Name of existing tool this validation wraps",
)
transform: str | None = Field(
default=None,
description="Python function code to transform wrapped tool output",
)
argument_mapping: dict[str, Any] | None = Field(
default=None,
description="Argument mapping for the wrapped tool (only valid with wraps)",
)
@model_validator(mode="after")
def _validate_wraps_fields(self) -> Validation:
"""Enforce wraps-related constraints."""
if self.wraps is not None:
# When wraps is set, source should be wrapped
if self.source != ToolSource.WRAPPED:
raise ValueError("source must be 'wrapped' when wraps is set")
if self.code is not None:
raise ValueError(
"code must not be set when wraps is set "
"(wraps and custom code are mutually exclusive)"
)
if self.transform is None:
raise ValueError("transform is required when wraps is set")
else:
# No wraps: argument_mapping is forbidden
if self.argument_mapping is not None:
raise ValueError("argument_mapping is only valid when wraps is set")
return self
@model_validator(mode="after")
def _enforce_validation_constraints(self) -> Validation:
"""Force validation-specific constraints on capability and tool_type."""
if self.tool_type != ToolType.VALIDATION:
object.__setattr__(self, "tool_type", ToolType.VALIDATION)
cap = self.capability
if cap.writes or cap.checkpointable or not cap.read_only:
forced = ToolCapability(
read_only=True,
writes=False,
write_scope=cap.write_scope,
checkpointable=False,
checkpoint_scope=cap.checkpoint_scope,
side_effects=cap.side_effects,
idempotent=cap.idempotent,
unsafe=cap.unsafe,
human_approval_required=cap.human_approval_required,
cost_profile=cap.cost_profile,
)
object.__setattr__(self, "capability", forced)
return self
@classmethod
def from_config(cls, config: dict[str, Any]) -> Validation:
"""Create a Validation from a YAML configuration dict.
Handles validation-specific fields (mode, wraps, transform,
argument_mapping) on top of the base Tool config.
"""
if "name" not in config:
raise ValueError("Validation config must include 'name'")
if "description" not in config:
raise ValueError("Validation config must include 'description'")
# Determine source: if wraps is set, source is implicitly 'wrapped'
if config.get("wraps"):
source = ToolSource.WRAPPED
elif "source" not in config:
raise ValueError("Validation config must include 'source' (or 'wraps')")
else:
source = ToolSource(config["source"])
# Build capability from nested dict
cap_data = config.get("capability", {})
capability = ToolCapability(**cap_data) if cap_data else ToolCapability()
# Build resource slots
slots_data = config.get("resource_slots", [])
resource_slots = [ResourceSlot(**s) for s in slots_data]
# Build lifecycle
lc_data = config.get("lifecycle")
lifecycle = ToolLifecycle(**lc_data) if lc_data else None
return cls(
name=config["name"],
description=config["description"],
source=source,
tool_type=ToolType.VALIDATION,
code=config.get("code"),
mcp_server=config.get("mcp_server"),
mcp_tool_name=config.get("mcp_tool_name"),
agent_skill_path=config.get("agent_skill_path"),
input_schema=config.get("input_schema"),
output_schema=config.get("output_schema"),
capability=capability,
resource_slots=resource_slots,
lifecycle=lifecycle,
timeout=config.get("timeout", 300),
mode=ValidationMode(config.get("mode", "required")),
wraps=config.get("wraps"),
transform=config.get("transform"),
argument_mapping=config.get("argument_mapping"),
)
def as_cli_dict(self) -> OrderedDict[str, Any]:
"""Return a stable-ordered dictionary for CLI rendering.
Extends the base Tool CLI dict with validation-specific fields.
"""
result = super().as_cli_dict()
result["mode"] = self.mode.value
if self.wraps:
result["wraps"] = self.wraps
if self.transform:
result["transform"] = self.transform
if self.argument_mapping:
result["argument_mapping"] = dict(self.argument_mapping)
return result
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
use_enum_values=False,
)