feat: Merge branch 'master' into feature/m1-resource-db-robot-tests

This commit is contained in:
2026-02-17 15:59:22 -08:00
24 changed files with 6678 additions and 42 deletions
@@ -0,0 +1,240 @@
"""Add tool registry tables (tools, tool_resource_bindings, validation_attachments).
This migration creates the three core tool registry tables needed for
the Tool Registry Persistence layer:
- ``tools``: Registered tool and validation definitions with namespaced
names as primary keys, JSON schema for inputs/outputs, capability
metadata, and source-specific fields.
- ``tool_resource_bindings``: Declared resource bindings per tool,
specifying typed resource slots with binding mode and access level.
- ``validation_attachments``: Runtime attachments linking validations to
resources with optional project/plan scoping and ULID primary keys.
Revision ID: c1_001_tool_registry
Revises: b1_001_resource_registry
Create Date: 2026-02-14 12:00:00
"""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "c1_001_tool_registry"
down_revision: str | Sequence[str] | None = "b0_001_projects"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Create tools, tool_resource_bindings, and validation_attachments tables."""
# --- tools table ---
op.create_table(
"tools",
# PK: namespaced tool name (e.g. "local/lint-check")
sa.Column("name", sa.String(255), nullable=False),
# Extracted namespace for filtering
sa.Column("namespace", sa.String(100), nullable=False),
# Extracted short name
sa.Column("short_name", sa.String(150), nullable=False),
# Human-readable description
sa.Column("description", sa.Text(), nullable=False),
# Discriminator: tool or validation
sa.Column("tool_type", sa.String(20), nullable=False),
# Source of implementation
sa.Column("source", sa.String(20), nullable=False),
# JSON Schema for inputs and outputs
sa.Column("input_schema_json", sa.Text(), nullable=True),
sa.Column("output_schema_json", sa.Text(), nullable=True),
# Capability metadata (JSON blob)
sa.Column("capability_json", sa.Text(), nullable=True),
# Lifecycle hooks (JSON blob)
sa.Column("lifecycle_json", sa.Text(), nullable=True),
# Inline code for source=custom
sa.Column("code", sa.Text(), nullable=True),
# MCP fields
sa.Column("mcp_server", sa.String(255), nullable=True),
sa.Column("mcp_tool_name", sa.String(255), nullable=True),
# Agent skill path
sa.Column("agent_skill_path", sa.String(1024), nullable=True),
# Execution timeout
sa.Column(
"timeout",
sa.Integer(),
nullable=False,
server_default=sa.text("300"),
),
# Validation wraps reference (FK to tools.name)
sa.Column(
"wraps",
sa.String(255),
sa.ForeignKey(
"tools.name",
ondelete="SET NULL",
name="fk_tools_wraps",
),
nullable=True,
),
# Transform code for wrapped validations
sa.Column("transform", sa.Text(), nullable=True),
# Validation mode (required/informational)
sa.Column("mode", sa.String(20), nullable=True),
# Argument mapping JSON for wrapped validations
sa.Column("argument_mapping_json", sa.Text(), nullable=True),
# Timestamps (ISO-8601 strings)
sa.Column("created_at", sa.String(30), nullable=False),
sa.Column("updated_at", sa.String(30), nullable=False),
# Constraints
sa.PrimaryKeyConstraint("name"),
sa.CheckConstraint(
"tool_type IN ('tool', 'validation')",
name="ck_tools_type",
),
sa.CheckConstraint(
"source IN ('mcp', 'agent_skill', 'builtin', 'custom', 'wrapped')",
name="ck_tools_source",
),
)
op.create_index(
"ix_tools_namespace",
"tools",
["namespace"],
unique=False,
)
op.create_index(
"ix_tools_tool_type",
"tools",
["tool_type"],
unique=False,
)
op.create_index(
"ix_tools_source",
"tools",
["source"],
unique=False,
)
# --- tool_resource_bindings table ---
op.create_table(
"tool_resource_bindings",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column(
"tool_name",
sa.String(255),
sa.ForeignKey(
"tools.name",
ondelete="CASCADE",
name="fk_tool_resource_bindings_tool",
),
nullable=False,
),
sa.Column("slot_name", sa.String(100), nullable=False),
sa.Column("resource_type", sa.String(255), nullable=False),
sa.Column("access_mode", sa.String(20), nullable=False),
sa.Column("binding_mode", sa.String(20), nullable=False),
sa.Column("static_resource", sa.String(255), nullable=True),
sa.Column(
"required",
sa.Boolean(),
nullable=False,
server_default=sa.text("1"),
),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("created_at", sa.String(30), nullable=False),
# Constraints
sa.PrimaryKeyConstraint("id"),
)
op.create_index(
"ix_tool_resource_bindings_tool_name",
"tool_resource_bindings",
["tool_name"],
unique=False,
)
# --- validation_attachments table ---
op.create_table(
"validation_attachments",
# PK: ULID (26-char string)
sa.Column("attachment_id", sa.String(26), nullable=False),
# FK to tools.name (must be a validation)
sa.Column(
"validation_name",
sa.String(255),
sa.ForeignKey(
"tools.name",
ondelete="CASCADE",
name="fk_validation_attachments_tool",
),
nullable=False,
),
# Resource reference (string identifier)
sa.Column("resource_id", sa.String(255), nullable=False),
# Mode: required or informational
sa.Column("mode", sa.String(20), nullable=False),
# Scoping fields
sa.Column("project_name", sa.String(255), nullable=True),
sa.Column("plan_id", sa.String(26), nullable=True),
# Optional arguments for the validation invocation
sa.Column("args_json", sa.Text(), nullable=True),
# Timestamp
sa.Column("created_at", sa.String(30), nullable=False),
# Constraints
sa.PrimaryKeyConstraint("attachment_id"),
sa.CheckConstraint(
"mode IN ('required', 'informational')",
name="ck_validation_attachments_mode",
),
)
op.create_index(
"ix_validation_attachments_validation_name",
"validation_attachments",
["validation_name"],
unique=False,
)
op.create_index(
"ix_validation_attachments_resource_id",
"validation_attachments",
["resource_id"],
unique=False,
)
op.create_index(
"ix_validation_attachments_project_name",
"validation_attachments",
["project_name"],
unique=False,
)
def downgrade() -> None:
"""Drop tool registry tables."""
# Drop in reverse FK order
op.drop_index(
"ix_validation_attachments_project_name",
table_name="validation_attachments",
)
op.drop_index(
"ix_validation_attachments_resource_id",
table_name="validation_attachments",
)
op.drop_index(
"ix_validation_attachments_validation_name",
table_name="validation_attachments",
)
op.drop_table("validation_attachments")
op.drop_index(
"ix_tool_resource_bindings_tool_name",
table_name="tool_resource_bindings",
)
op.drop_table("tool_resource_bindings")
op.drop_index("ix_tools_source", table_name="tools")
op.drop_index("ix_tools_tool_type", table_name="tools")
op.drop_index("ix_tools_namespace", table_name="tools")
op.drop_table("tools")
+358
View File
@@ -0,0 +1,358 @@
"""ASV benchmarks for Actor YAML schema validation throughput.
Measures the performance of:
- YAML string parsing + schema validation
- YAML file loading + schema validation
- Graph topology validation (cycle detection)
- Tool definition validation
- model_dump() serialization
"""
from __future__ import annotations
import importlib
import sys
import tempfile
from pathlib import Path
# Ensure the local *source* tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Force-reload the top-level package
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.actor.schema import ActorConfigSchema # noqa: E402
_MINIMAL_LLM_YAML = """\
name: bench/llm
type: llm
description: Benchmark LLM actor
model: gpt-4
"""
_FULL_LLM_YAML = """\
name: bench/llm_full
type: llm
description: Full LLM actor with all features
version: "1.0"
model: gpt-4-turbo
system_prompt: |
You are a benchmark testing assistant.
Process requests efficiently and accurately.
tools:
- files/read_file
- files/write_file
context_view: executor
memory:
enabled: true
max_messages: 100
max_tokens: 8000
summarize_old: true
context:
include_files:
- README.md
- pyproject.toml
include_dirs:
- src/
- tests/
exclude_patterns:
- "**/__pycache__/**"
- "*.pyc"
max_context_tokens: 16000
env_vars:
LOG_LEVEL: info
WORK_DIR: /tmp/bench
"""
_TOOL_ACTOR_YAML = """\
name: bench/tools
type: tool
description: Tool collection for benchmarking
tools:
- files/read_file
- files/write_file
- files/delete_file
"""
_SIMPLE_GRAPH_YAML = """\
name: bench/simple_graph
type: graph
description: Simple 3-node graph for benchmarking
model: gpt-4
route:
nodes:
- id: extract
type: tool
name: Extractor
description: Extract data
config:
tool_name: data/extract
- id: process
type: agent
name: Processor
description: Process data
config:
prompt: "Process the data"
- id: save
type: tool
name: Saver
description: Save results
config:
tool_name: data/save
edges:
- from_node: extract
to_node: process
- from_node: process
to_node: save
entry_node: extract
exit_nodes:
- save
"""
_COMPLEX_GRAPH_YAML = """\
name: bench/complex_graph
type: graph
description: Complex graph with 10 nodes and conditionals
model: gpt-4
route:
nodes:
- id: start
type: agent
name: Starter
description: Start node
config:
prompt: "Begin"
- id: node2
type: agent
name: Node 2
description: Second node
config:
prompt: "Continue"
- id: node3
type: agent
name: Node 3
description: Third node
config:
prompt: "Process"
- id: checker
type: conditional
name: Checker
description: Check condition
config:
conditions:
- check: "state.get('ok') == True"
route_to: node4
- check: "state.get('ok') == False"
route_to: node5
- id: node4
type: agent
name: Node 4
description: Success path
config:
prompt: "Success"
- id: node5
type: agent
name: Node 5
description: Retry path
config:
prompt: "Retry"
- id: node6
type: tool
name: Tool 6
description: Tool execution
config:
tool_name: test/tool
- id: node7
type: agent
name: Node 7
description: Seventh node
config:
prompt: "Continue"
- id: node8
type: subgraph
name: Subgraph
description: Nested workflow
config:
actor_path: bench/nested.yaml
- id: end
type: agent
name: End
description: Final node
config:
prompt: "Complete"
edges:
- from_node: start
to_node: node2
- from_node: node2
to_node: node3
- from_node: node3
to_node: checker
- from_node: node4
to_node: node6
- from_node: node5
to_node: node7
- from_node: node6
to_node: node8
- from_node: node7
to_node: node8
- from_node: node8
to_node: end
entry_node: start
exit_nodes:
- end
"""
class TimeActorSchemaMinimalLLM:
"""Benchmark minimal LLM actor validation."""
def time_parse_minimal_llm(self) -> None:
"""Time parsing a minimal LLM actor YAML."""
import yaml
data = yaml.safe_load(_MINIMAL_LLM_YAML)
ActorConfigSchema.model_validate(data)
class TimeActorSchemaFullLLM:
"""Benchmark full LLM actor validation."""
def time_parse_full_llm(self) -> None:
"""Time parsing a full LLM actor YAML with all features."""
import yaml
data = yaml.safe_load(_FULL_LLM_YAML)
ActorConfigSchema.model_validate(data)
class TimeActorSchemaTool:
"""Benchmark tool actor validation."""
def time_parse_tool_actor(self) -> None:
"""Time parsing a tool actor YAML."""
import yaml
data = yaml.safe_load(_TOOL_ACTOR_YAML)
ActorConfigSchema.model_validate(data)
class TimeActorSchemaSimpleGraph:
"""Benchmark simple graph actor validation."""
def time_parse_simple_graph(self) -> None:
"""Time parsing a simple 3-node graph actor."""
import yaml
data = yaml.safe_load(_SIMPLE_GRAPH_YAML)
ActorConfigSchema.model_validate(data)
class TimeActorSchemaComplexGraph:
"""Benchmark complex graph actor validation."""
def time_parse_complex_graph(self) -> None:
"""Time parsing a complex 10-node graph with conditionals."""
import yaml
data = yaml.safe_load(_COMPLEX_GRAPH_YAML)
ActorConfigSchema.model_validate(data)
class TimeActorSchemaFileIO:
"""Benchmark file I/O operations."""
def setup(self) -> None:
"""Create temporary YAML file."""
with tempfile.NamedTemporaryFile(
mode="w", suffix=".yaml", delete=False
) as temp_file:
temp_file.write(_FULL_LLM_YAML)
self.temp_file_name = temp_file.name
def teardown(self) -> None:
"""Remove temporary file."""
Path(self.temp_file_name).unlink(missing_ok=True)
def time_load_from_file(self) -> None:
"""Time loading actor config from YAML file."""
ActorConfigSchema.from_yaml_file(self.temp_file_name)
def time_save_to_file(self) -> None:
"""Time saving actor config to YAML file."""
import yaml
data = yaml.safe_load(_FULL_LLM_YAML)
config = ActorConfigSchema.model_validate(data)
with tempfile.NamedTemporaryFile(
mode="w", suffix=".yaml", delete=False
) as temp_out:
temp_out_name = temp_out.name
try:
config.to_yaml_file(temp_out_name)
finally:
Path(temp_out_name).unlink(missing_ok=True)
class TimeActorSchemaSerialization:
"""Benchmark serialization operations."""
def setup(self) -> None:
"""Parse actor configs once."""
import yaml
self.minimal_config = ActorConfigSchema.model_validate(
yaml.safe_load(_MINIMAL_LLM_YAML)
)
self.full_config = ActorConfigSchema.model_validate(
yaml.safe_load(_FULL_LLM_YAML)
)
self.graph_config = ActorConfigSchema.model_validate(
yaml.safe_load(_COMPLEX_GRAPH_YAML)
)
def time_dump_minimal(self) -> None:
"""Time model_dump() on minimal config."""
self.minimal_config.model_dump(mode="json")
def time_dump_full(self) -> None:
"""Time model_dump() on full config."""
self.full_config.model_dump(mode="json")
def time_dump_graph(self) -> None:
"""Time model_dump() on complex graph."""
self.graph_config.model_dump(mode="json")
class TimeActorSchemaGraphValidation:
"""Benchmark graph topology validation."""
def setup(self) -> None:
"""Parse graph configs."""
import yaml
self.simple_graph_data = yaml.safe_load(_SIMPLE_GRAPH_YAML)
self.complex_graph_data = yaml.safe_load(_COMPLEX_GRAPH_YAML)
def time_validate_simple_graph_topology(self) -> None:
"""Time validation of simple graph topology."""
ActorConfigSchema.model_validate(self.simple_graph_data)
def time_validate_complex_graph_topology(self) -> None:
"""Time validation of complex graph with cycle detection."""
ActorConfigSchema.model_validate(self.complex_graph_data)
# Module-level metadata for ASV
time_parse_minimal_llm = TimeActorSchemaMinimalLLM()
time_parse_full_llm = TimeActorSchemaFullLLM()
time_parse_tool_actor = TimeActorSchemaTool()
time_parse_simple_graph = TimeActorSchemaSimpleGraph()
time_parse_complex_graph = TimeActorSchemaComplexGraph()
time_file_io = TimeActorSchemaFileIO()
time_serialization = TimeActorSchemaSerialization()
time_graph_validation = TimeActorSchemaGraphValidation()
+162
View File
@@ -0,0 +1,162 @@
"""ASV benchmarks for Tool Registry persistence operations.
Measures the performance of:
- ToolModel.from_domain() construction
- ToolModel.to_domain() reconstruction
- ToolRegistryRepository CRUD operations (in-memory SQLite)
- ValidationAttachmentRepository attach/list operations
"""
from __future__ import annotations
import sys
from datetime import datetime
from pathlib import Path
try:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.infrastructure.database.models import Base, ToolModel
from cleveragents.infrastructure.database.repositories import (
ToolRegistryRepository,
ValidationAttachmentRepository,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.infrastructure.database.models import Base, ToolModel
from cleveragents.infrastructure.database.repositories import (
ToolRegistryRepository,
ValidationAttachmentRepository,
)
def _make_tool_dict(
name: str = "bench/tool-test",
source: str = "builtin",
tool_type: str = "tool",
) -> dict[str, object]:
"""Create a minimal tool dict for benchmarks."""
now = datetime.now().isoformat()
return {
"name": name,
"description": f"Benchmark tool {name}",
"tool_type": tool_type,
"source": source,
"timeout": 300,
"created_at": now,
"updated_at": now,
"resource_bindings": [
{
"slot_name": "repo",
"resource_type": "git-checkout",
"access_mode": "read_only",
"binding_mode": "contextual",
"required": True,
},
],
}
def _setup_db():
"""Create in-memory DB and return (session, tool_repo, att_repo)."""
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, expire_on_commit=False)
session = factory()
tool_repo = ToolRegistryRepository(session_factory=lambda: session)
att_repo = ValidationAttachmentRepository(session_factory=lambda: session)
return session, tool_repo, att_repo
class ToolModelSuite:
"""Benchmark ToolModel ORM construction and conversion."""
def setup(self) -> None:
"""Prepare objects."""
self.tool_dict = _make_tool_dict()
def time_from_domain(self) -> None:
"""Benchmark ToolModel.from_domain() construction."""
ToolModel.from_domain(self.tool_dict)
def time_to_domain(self) -> None:
"""Benchmark ToolModel.to_domain() conversion."""
model = ToolModel.from_domain(self.tool_dict)
model.resource_bindings_rel = []
model.to_domain()
class ToolRegistryCRUDSuite:
"""Benchmark ToolRegistryRepository CRUD operations."""
def setup(self) -> None:
"""Prepare a fresh in-memory database for each benchmark."""
self.session, self.repo, self.att_repo = _setup_db()
self._counter = 0
def time_create_tool(self) -> None:
"""Benchmark creating a single tool."""
self._counter += 1
tool = _make_tool_dict(name=f"bench/tool-{self._counter}")
self.repo.create(tool)
self.session.commit()
def time_get_by_name(self) -> None:
"""Benchmark retrieving a tool by name."""
name = "bench/lookup-target"
if self.repo.get_by_name(name) is None:
self.repo.create(_make_tool_dict(name=name))
self.session.commit()
self.repo.get_by_name(name)
def time_list_all(self) -> None:
"""Benchmark listing all tools."""
# Ensure some data exists
for i in range(5):
n = f"bench/list-{i}"
if self.repo.get_by_name(n) is None:
self.repo.create(_make_tool_dict(name=n))
self.session.commit()
self.repo.list_all()
class ValidationAttachmentSuite:
"""Benchmark ValidationAttachmentRepository operations."""
def setup(self) -> None:
"""Prepare database with a validation tool."""
self.session, self.repo, self.att_repo = _setup_db()
val = _make_tool_dict(
name="bench/val-bench",
tool_type="validation",
)
val["mode"] = "required"
self.repo.create(val)
self.session.commit()
self._counter = 0
def time_attach(self) -> None:
"""Benchmark attaching a validation."""
self._counter += 1
self.att_repo.attach(
validation_name="bench/val-bench",
resource_id=f"res-{self._counter}",
mode="required",
)
self.session.commit()
def time_list_for_resource(self) -> None:
"""Benchmark listing attachments for a resource."""
res_id = "res-bench-list"
if not self.att_repo.list_for_resource(res_id):
self.att_repo.attach(
validation_name="bench/val-bench",
resource_id=res_id,
mode="required",
)
self.session.commit()
self.att_repo.list_for_resource(res_id)
File diff suppressed because it is too large Load Diff
+215
View File
@@ -0,0 +1,215 @@
# Graph Actor - Test-Driven Development Workflow
# Demonstrates a multi-node graph with conditional routing and subgraphs
name: workflows/test_driven_dev
type: graph
description: Test-driven development workflow with automated testing and feedback loops
version: "1.0"
# LLM model for agent nodes
model: gpt-4
# Graph topology
route:
# Define all nodes in the workflow
nodes:
# Entry point: planning agent
- id: planner
type: agent
name: Test Planner
description: Plans test cases based on requirements
config:
model: gpt-4
prompt: |
You are a test planning expert. Analyze the requirements and create
a comprehensive test plan covering:
- Unit tests for individual functions
- Integration tests for component interactions
- Edge cases and error conditions
tools:
- files/read_file
- files/list_directory
# Write tests first
- id: test_writer
type: agent
name: Test Writer
description: Writes test cases based on the plan
config:
model: gpt-4
prompt: |
You are a test writing expert. Write pytest tests based on the plan.
Follow best practices:
- Use descriptive test names
- Include docstrings
- Use fixtures appropriately
- Test one thing per test
tools:
- files/write_file
- files/read_file
# Run the tests (should fail initially)
- id: run_tests
type: tool
name: Test Runner
description: Executes pytest test suite
config:
tool_name: testing/run_pytest
parameters:
verbose: true
coverage: true
# Check test results
- id: check_results
type: conditional
name: Test Result Checker
description: Routes based on test pass/fail status
config:
conditions:
- check: "state.get('tests_passed') == True"
route_to: code_review
- check: "state.get('tests_passed') == False"
route_to: implementation_writer
# Write implementation to make tests pass
- id: implementation_writer
type: agent
name: Implementation Writer
description: Writes code to make the tests pass
config:
model: gpt-4
prompt: |
You are an implementation expert. Write clean, well-documented code
that makes the failing tests pass. Follow SOLID principles and
write maintainable code.
tools:
- files/write_file
- files/read_file
# Run tests again after implementation
- id: rerun_tests
type: tool
name: Test Rerunner
description: Re-executes tests after implementation
config:
tool_name: testing/run_pytest
parameters:
verbose: true
coverage: true
# Check if tests pass now
- id: verify_tests
type: conditional
name: Test Verification
description: Verify tests pass after implementation
config:
conditions:
- check: "state.get('tests_passed') == True"
route_to: code_review
- check: "state.get('tests_passed') == False and state.get('retry_count', 0) < 3"
route_to: debug_failures
- check: "state.get('tests_passed') == False and state.get('retry_count', 0) >= 3"
route_to: escalate
# Debug test failures
- id: debug_failures
type: agent
name: Debugger
description: Analyzes and fixes test failures
config:
model: gpt-4
prompt: |
You are a debugging expert. Analyze the test failures and fix the
implementation. Look for:
- Logic errors
- Edge cases
- Type mismatches
- Missing error handling
tools:
- files/read_file
- files/write_file
# Code review (subgraph)
- id: code_review
type: subgraph
name: Code Reviewer
description: Runs code review workflow
config:
actor_path: examples/actors/simple_llm.yaml
# Escalate if tests keep failing
- id: escalate
type: tool
name: Escalation Handler
description: Escalates persistent failures to human
config:
tool_name: notifications/send_alert
parameters:
channel: engineering
priority: high
# Define edges (workflow transitions)
edges:
# Linear flow from planner to test writer
- from_node: planner
to_node: test_writer
# Run tests after writing them
- from_node: test_writer
to_node: run_tests
# Check results after running tests
- from_node: run_tests
to_node: check_results
# Conditional routing from check_results
# (handled by the conditional node itself)
# Write implementation if tests fail
- from_node: implementation_writer
to_node: rerun_tests
# Verify tests after rerunning
- from_node: rerun_tests
to_node: verify_tests
# Debug if tests still fail
- from_node: debug_failures
to_node: rerun_tests
# All paths eventually lead to code review or escalation
# (handled by conditional nodes)
# Entry and exit points
entry_node: planner
exit_nodes:
- code_review
- escalate
# Context settings
context_view: strategist
memory:
enabled: true
max_messages: 100
max_tokens: 16000
summarize_old: true
context:
include_files:
- "README.md"
- "requirements.txt"
- "pyproject.toml"
include_dirs:
- "src/"
- "tests/"
exclude_patterns:
- "**/__pycache__/**"
- "*.pyc"
- "**/.pytest_cache/**"
- "**/htmlcov/**"
max_context_tokens: 32000
# Environment variables
env_vars:
PYTEST_ARGS: --verbose --cov --cov-report=html
MAX_RETRIES: "3"
+67
View File
@@ -0,0 +1,67 @@
# LLM Actor with Tools
# Demonstrates an LLM actor with access to multiple tools
name: assistants/file_analyzer
type: llm
description: Analyzes files and generates reports using file system tools
version: "1.0"
# LLM configuration
model: gpt-4-turbo
system_prompt: |
You are a file analysis assistant. Use the available tools to:
- Read and analyze file contents
- Count lines, words, and characters
- Search for patterns in files
- Generate summary reports
Always explain what you're doing before using a tool.
# Tools (mix of references and inline definitions)
tools:
# Reference to existing tool
- files/read_file
- files/list_directory
# Inline tool definition
- name: utils/count_lines
description: Count the number of lines in a file
parameters:
- name: file_path
type: str
description: Path to the file to count lines in
required: true
code: |
def count_lines(file_path: str) -> int:
"""Count lines in a file."""
with open(file_path, 'r', encoding='utf-8') as f:
return len(f.readlines())
- name: utils/word_count
description: Count words in a file
parameters:
- name: file_path
type: str
description: Path to the file
required: true
code: |
def word_count(file_path: str) -> int:
"""Count words in a file."""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return len(content.split())
# Context settings
context_view: executor
memory:
enabled: true
max_messages: 30
max_tokens: 6000
context:
max_context_tokens: 10000
# Environment variables
env_vars:
WORK_DIR: ${HOME}/workspace
LOG_LEVEL: info
+78
View File
@@ -0,0 +1,78 @@
# Simple Graph Actor - Sequential Processing
# Demonstrates a simple 3-node graph with linear execution
name: workflows/document_processor
type: graph
description: Simple document processing workflow (extract → analyze → summarize)
version: "1.0"
# LLM model
model: gpt-3.5-turbo
# Graph topology
route:
nodes:
# Node 1: Extract text from document
- id: extractor
type: tool
name: Text Extractor
description: Extracts text from various document formats
config:
tool_name: documents/extract_text
parameters:
formats:
- pdf
- docx
- txt
# Node 2: Analyze content
- id: analyzer
type: agent
name: Content Analyzer
description: Analyzes document structure and content
config:
model: gpt-3.5-turbo
prompt: |
Analyze the document content and identify:
- Main topics and themes
- Key entities (people, places, organizations)
- Sentiment and tone
- Document structure
tools:
- analysis/extract_entities
- analysis/sentiment_analysis
# Node 3: Generate summary
- id: summarizer
type: agent
name: Summarizer
description: Creates concise summary of document
config:
model: gpt-3.5-turbo
prompt: |
Create a concise summary of the document including:
- Main points (3-5 bullet points)
- Key findings
- Actionable insights
Keep it under 200 words.
# Linear edges
edges:
- from_node: extractor
to_node: analyzer
- from_node: analyzer
to_node: summarizer
# Entry and exit
entry_node: extractor
exit_nodes:
- summarizer
# Context settings
context_view: executor
memory:
enabled: true
max_messages: 10
context:
max_context_tokens: 4000
+38
View File
@@ -0,0 +1,38 @@
# Simple LLM Actor Example
# Demonstrates the most basic actor configuration with just an LLM and system prompt
name: assistants/code_reviewer
type: llm
description: Reviews Python code for best practices, style, and potential bugs
version: "1.0"
# LLM configuration
model: gpt-4
system_prompt: |
You are an expert Python code reviewer. Review code for:
- PEP 8 style compliance
- Best practices and design patterns
- Potential bugs and edge cases
- Performance considerations
- Security vulnerabilities
Provide constructive feedback with specific suggestions for improvement.
# Context and memory settings
context_view: reviewer
memory:
enabled: true
max_messages: 20
max_tokens: 4000
context:
include_files:
- "README.md"
- "pyproject.toml"
include_dirs:
- "src/"
exclude_patterns:
- "**/__pycache__/**"
- "*.pyc"
- "**/.pytest_cache/**"
max_context_tokens: 8000
+45
View File
@@ -0,0 +1,45 @@
# Tool-Only Actor
# Demonstrates an actor that only provides tools without LLM interaction
name: utilities/file_operations
type: tool
description: Collection of file operation tools for other actors to use
version: "1.0"
# Tool collection
tools:
# Reference existing tools
- files/read_file
- files/write_file
- files/delete_file
- files/copy_file
- files/move_file
- files/list_directory
- files/create_directory
# Add custom validation tool
- name: validators/check_python_syntax
description: Validate Python file syntax without executing it
parameters:
- name: file_path
type: str
description: Path to Python file to validate
required: true
code: |
import ast
def check_python_syntax(file_path: str) -> dict:
"""Check if a Python file has valid syntax."""
try:
with open(file_path, 'r', encoding='utf-8') as f:
ast.parse(f.read())
return {"valid": True, "error": None}
except SyntaxError as e:
return {
"valid": False,
"error": str(e),
"line": e.lineno,
"offset": e.offset
}
# No LLM configuration needed for tool-only actors
# No system prompt needed
+355
View File
@@ -0,0 +1,355 @@
Feature: Actor YAML schema validation
As a CleverAgents v3 user
I want to define actors via YAML configuration files
So that I can create validated, reusable actor workflows
#
# Valid LLM Actor scenarios
#
Scenario: Load a minimal valid LLM actor
Given an actor YAML string with minimal LLM configuration
When I validate the actor schema
Then the actor schema validation should succeed
And the actor config name should be "assistants/simple"
And the actor config type should be "llm"
And the actor config model should be "gpt-4"
Scenario: Load LLM actor with system prompt
Given an actor YAML string with LLM and system prompt
When I validate the actor schema
Then the actor schema validation should succeed
And the actor config system_prompt should contain "expert"
Scenario: Load LLM actor with tools
Given an actor YAML string with LLM and tools
When I validate the actor schema
Then the actor schema validation should succeed
And the actor config should have 2 tools
Scenario: Load LLM actor with memory config
Given an actor YAML string with memory configuration
When I validate the actor schema
Then the actor schema validation should succeed
And the actor memory enabled should be true
And the actor memory max_messages should be 50
Scenario: Load LLM actor with context config
Given an actor YAML string with context configuration
When I validate the actor schema
Then the actor schema validation should succeed
And the actor context should include 2 files
And the actor context should include 1 directory
Scenario: Load simple_llm.yaml example
Given the actor YAML file "examples/actors/simple_llm.yaml"
When I validate the actor schema from file
Then the actor schema validation should succeed
And the actor config type should be "llm"
Scenario: Load llm_with_tools.yaml example
Given the actor YAML file "examples/actors/llm_with_tools.yaml"
When I validate the actor schema from file
Then the actor schema validation should succeed
And the actor config should have at least 2 tools
# ────────────────────────────────────────────────────────────
# Valid TOOL Actor scenarios
# ────────────────────────────────────────────────────────────
Scenario: Load a minimal valid TOOL actor
Given an actor YAML string with TOOL type and tools
When I validate the actor schema
Then the actor schema validation should succeed
And the actor config type should be "tool"
And the actor config should have at least 1 tool
Scenario: Load TOOL actor with inline tool definition
Given an actor YAML string with inline tool definition
When I validate the actor schema
Then the actor schema validation should succeed
And the actor config should have 1 inline tool
Scenario: Load tool_collection.yaml example
Given the actor YAML file "examples/actors/tool_collection.yaml"
When I validate the actor schema from file
Then the actor schema validation should succeed
And the actor config type should be "tool"
# ────────────────────────────────────────────────────────────
# Valid GRAPH Actor scenarios
# ────────────────────────────────────────────────────────────
Scenario: Load a minimal valid GRAPH actor
Given an actor YAML string with minimal GRAPH configuration
When I validate the actor schema
Then the actor schema validation should succeed
And the actor config type should be "graph"
And the actor route should have entry_node "start"
Scenario: Load GRAPH with linear topology
Given an actor YAML string with linear graph topology
When I validate the actor schema
Then the actor schema validation should succeed
And the actor route should have 3 nodes
And the actor route should have 2 edges
Scenario: Load GRAPH with conditional routing
Given an actor YAML string with conditional routing
When I validate the actor schema
Then the actor schema validation should succeed
And the actor route should have 1 conditional node
Scenario: Load GRAPH with subgraph node
Given an actor YAML string with subgraph node
When I validate the actor schema
Then the actor schema validation should succeed
And the actor route should have 1 subgraph node
Scenario: Load simple_graph.yaml example
Given the actor YAML file "examples/actors/simple_graph.yaml"
When I validate the actor schema from file
Then the actor schema validation should succeed
And the actor config type should be "graph"
Scenario: Load graph_workflow.yaml example
Given the actor YAML file "examples/actors/graph_workflow.yaml"
When I validate the actor schema from file
Then the actor schema validation should succeed
And the actor config type should be "graph"
# ────────────────────────────────────────────────────────────
# Invalid Actor Name scenarios
# ────────────────────────────────────────────────────────────
Scenario: Reject actor without namespace
Given an actor YAML string with name "simple_actor"
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "namespaced"
Scenario: Reject actor with multiple slashes
Given an actor YAML string with name "namespace/sub/actor"
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "namespaced"
Scenario: Reject actor with empty namespace
Given an actor YAML string with name "/actor"
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "namespaced"
Scenario: Reject actor with empty name
Given an actor YAML string with name "namespace/"
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "namespaced"
# ────────────────────────────────────────────────────────────
# Invalid LLM Actor scenarios
# ────────────────────────────────────────────────────────────
Scenario: Reject LLM actor without model
Given an actor YAML string with LLM type but no model
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "require 'model'"
# ────────────────────────────────────────────────────────────
# Invalid TOOL Actor scenarios
# ────────────────────────────────────────────────────────────
Scenario: Reject TOOL actor without tools
Given an actor YAML string with TOOL type but no tools
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "require at least one tool"
Scenario: Reject TOOL actor with empty tools list
Given an actor YAML string with TOOL type and empty tools
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "require at least one tool"
# ────────────────────────────────────────────────────────────
# Invalid GRAPH Actor scenarios
# ────────────────────────────────────────────────────────────
Scenario: Reject GRAPH actor without model
Given an actor YAML string with GRAPH type but no model
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "require 'model'"
Scenario: Reject GRAPH actor without route
Given an actor YAML string with GRAPH type but no route
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "require 'route'"
# ────────────────────────────────────────────────────────────
# Graph Topology Validation scenarios
# ────────────────────────────────────────────────────────────
Scenario: Reject graph with duplicate node IDs
Given an actor YAML string with duplicate node IDs
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "Duplicate node IDs"
Scenario: Reject graph with invalid entry node
Given an actor YAML string with non-existent entry node
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "Entry node"
And the validation error should contain "not found"
Scenario: Reject graph with invalid exit node
Given an actor YAML string with non-existent exit node
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "Exit node"
And the validation error should contain "not found"
Scenario: Reject graph with invalid edge from_node
Given an actor YAML string with invalid edge from_node
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "from_node"
And the validation error should contain "not found"
Scenario: Reject graph with invalid edge to_node
Given an actor YAML string with invalid edge to_node
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "to_node"
And the validation error should contain "not found"
Scenario: Reject graph with cycles
Given an actor YAML string with cyclic graph
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "cycle"
# ────────────────────────────────────────────────────────────
# Tool Definition Validation scenarios
# ────────────────────────────────────────────────────────────
Scenario: Reject inline tool without namespace
Given an actor YAML string with inline tool name "count_lines"
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "namespaced"
Scenario: Reject inline tool with invalid parameter name
Given an actor YAML string with invalid tool parameter name
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "valid Python identifier"
# ────────────────────────────────────────────────────────────
# Node Definition Validation scenarios
# ────────────────────────────────────────────────────────────
Scenario: Reject node with invalid ID format
Given an actor YAML string with node ID containing spaces
When I validate the actor schema
Then the actor schema validation should fail
And the validation error should contain "alphanumeric"
Scenario: Accept node with underscores and hyphens
Given an actor YAML string with node ID "node_1-test"
When I validate the actor schema
Then the actor schema validation should succeed
# ────────────────────────────────────────────────────────────
# Context View scenarios
# ────────────────────────────────────────────────────────────
Scenario: Load actor with strategist context view
Given an actor YAML string with context_view "strategist"
When I validate the actor schema
Then the actor schema validation should succeed
And the actor context_view should be "strategist"
Scenario: Load actor with executor context view
Given an actor YAML string with context_view "executor"
When I validate the actor schema
Then the actor schema validation should succeed
And the actor context_view should be "executor"
Scenario: Load actor with reviewer context view
Given an actor YAML string with context_view "reviewer"
When I validate the actor schema
Then the actor schema validation should succeed
And the actor context_view should be "reviewer"
Scenario: Load actor with full context view
Given an actor YAML string with context_view "full"
When I validate the actor schema
Then the actor schema validation should succeed
And the actor context_view should be "full"
# ────────────────────────────────────────────────────────────
# Environment Variables scenarios
# ────────────────────────────────────────────────────────────
Scenario: Load actor with environment variables
Given an actor YAML string with env_vars
When I validate the actor schema
Then the actor schema validation should succeed
And the actor should have 2 env_vars
# ────────────────────────────────────────────────────────────
# YAML I/O scenarios
# ────────────────────────────────────────────────────────────
Scenario: Save and reload actor configuration
Given a valid actor configuration object
When I save the actor to YAML file
And I reload the actor from YAML file
Then the reloaded actor should match the original
Scenario: Handle missing YAML file
Given a non-existent actor YAML file path
When I attempt to load the actor from file
Then a FileNotFoundError should be raised
# ────────────────────────────────────────────────────────────
# Edge Priority scenarios
# ────────────────────────────────────────────────────────────
Scenario: Load graph with edge priorities
Given an actor YAML string with edges having different priorities
When I validate the actor schema
Then the actor schema validation should succeed
And the highest priority edge should be 10
# ────────────────────────────────────────────────────────────
# Memory Configuration scenarios
# ────────────────────────────────────────────────────────────
Scenario: Load actor with memory disabled
Given an actor YAML string with memory enabled false
When I validate the actor schema
Then the actor schema validation should succeed
And the actor memory enabled should be false
Scenario: Load actor with message limit
Given an actor YAML string with max_messages 100
When I validate the actor schema
Then the actor schema validation should succeed
And the actor memory max_messages should be 100
Scenario: Load actor with token limit
Given an actor YAML string with max_tokens 8000
When I validate the actor schema
Then the actor schema validation should succeed
And the actor memory max_tokens should be 8000
Scenario: Load actor with summarize_old enabled
Given an actor YAML string with summarize_old true
When I validate the actor schema
Then the actor schema validation should succeed
And the actor memory summarize_old should be true
File diff suppressed because it is too large Load Diff
+597
View File
@@ -0,0 +1,597 @@
"""Step definitions for Tool Registry persistence coverage.
Covers tools, tool_resource_bindings, validation_attachments tables
and the ToolRegistryService integration layer.
"""
from __future__ import annotations
from datetime import datetime
from behave import given, then, when
from behave.runner import Context
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from cleveragents.application.services.tool_registry_service import (
ToolRegistryService,
)
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
DuplicateToolError,
ToolInUseError,
ToolRegistryRepository,
ValidationAttachmentRepository,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_tool_dict(
name: str = "core/test-tool",
source: str = "builtin",
tool_type: str = "tool",
description: str | None = None,
code: str | None = None,
mcp_server: str | None = None,
mcp_tool_name: str | None = None,
mode: str | None = None,
resource_bindings: list[dict[str, object]] | None = None,
) -> dict[str, object]:
"""Create a minimal valid tool dict for testing."""
now_iso = datetime.now().isoformat()
result: dict[str, object] = {
"name": name,
"description": description or f"Test tool {name}",
"tool_type": tool_type,
"source": source,
"timeout": 300,
"created_at": now_iso,
"updated_at": now_iso,
"resource_bindings": resource_bindings or [],
}
if code is not None:
result["code"] = code
if mcp_server is not None:
result["mcp_server"] = mcp_server
if mcp_tool_name is not None:
result["mcp_tool_name"] = mcp_tool_name
if mode is not None:
result["mode"] = mode
return result
def _make_binding(slot_name: str = "repo") -> dict[str, object]:
"""Create a resource binding dict."""
return {
"slot_name": slot_name,
"resource_type": "git-checkout",
"access_mode": "read_only",
"binding_mode": "contextual",
"required": True,
}
def _get_session_factory(context: Context) -> sessionmaker[Session]:
return context.tool_session_factory # type: ignore[no-any-return]
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a clean in-memory database with the tool registry schema")
def step_clean_tool_db(context: Context) -> None:
"""Set up an in-memory SQLite database with all tables."""
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
factory: sessionmaker[Session] = sessionmaker(bind=engine, expire_on_commit=False)
context.tool_engine = engine
context.tool_session_factory = factory
# Keep a persistent session for attachment repo (needs same connection)
context.tool_session = factory()
@given("a tool registry repository backed by the database")
def step_tool_repo(context: Context) -> None:
"""Create a ToolRegistryRepository."""
context.tool_repo = ToolRegistryRepository(
session_factory=lambda: context.tool_session,
)
@given("a validation attachment repository backed by the database")
def step_attachment_repo(context: Context) -> None:
"""Create a ValidationAttachmentRepository."""
context.attachment_repo = ValidationAttachmentRepository(
session_factory=lambda: context.tool_session,
)
# ---------------------------------------------------------------------------
# Given: tool dicts
# ---------------------------------------------------------------------------
@given('a valid tool dict named "{name}" with source "{source}"')
def step_valid_tool_dict(context: Context, name: str, source: str) -> None:
"""Build a tool dict."""
context.tool_dict = _make_tool_dict(name=name, source=source)
@given('the tool dict includes inline code "{code}"')
def step_tool_dict_code(context: Context, code: str) -> None:
context.tool_dict["code"] = code
@given('the tool dict includes mcp_server "{server}" and mcp_tool_name "{tool_name}"')
def step_tool_dict_mcp(context: Context, server: str, tool_name: str) -> None:
context.tool_dict["mcp_server"] = server
context.tool_dict["mcp_tool_name"] = tool_name
@given('the tool dict has tool_type "{tool_type}" and mode "{mode}"')
def step_tool_dict_type_mode(context: Context, tool_type: str, mode: str) -> None:
context.tool_dict["tool_type"] = tool_type
context.tool_dict["mode"] = mode
@given('the tool dict includes a resource binding for slot "{slot_name}"')
def step_tool_dict_binding(context: Context, slot_name: str) -> None:
bindings = list(context.tool_dict.get("resource_bindings", [])) # type: ignore[union-attr]
bindings.append(_make_binding(slot_name))
context.tool_dict["resource_bindings"] = bindings
@given("the tool has already been registered once in the tool registry")
def step_tool_already_registered(context: Context) -> None:
context.tool_repo.create(context.tool_dict)
context.tool_session.commit()
@given("the tool has been registered through the tool registry repository")
def step_tool_registered(context: Context) -> None:
context.tool_repo.create(context.tool_dict)
context.tool_session.commit()
@given("the following tools have been registered:")
def step_register_multiple_tools(context: Context) -> None:
for row in context.table:
tool_dict = _make_tool_dict(
name=row["name"],
source=row["source"],
tool_type=row["tool_type"],
)
context.tool_repo.create(tool_dict)
context.tool_session.commit()
@given('an unscoped validation attachment for "{val_name}" on resource "{res_id}"')
def step_attachment_exists(context: Context, val_name: str, res_id: str) -> None:
att = context.attachment_repo.attach(
validation_name=val_name,
resource_id=res_id,
mode="required",
)
context.tool_session.commit()
context.last_attachment = att
@given(
'a project-scoped validation attachment for "{val_name}" on resource "{res_id}" in project "{project}"'
)
def step_attachment_in_project(
context: Context, val_name: str, res_id: str, project: str
) -> None:
att = context.attachment_repo.attach(
validation_name=val_name,
resource_id=res_id,
mode="required",
project_name=project,
)
context.tool_session.commit()
context.last_attachment = att
@given(
'a plan-scoped validation attachment for "{val_name}" on resource "{res_id}" in plan "{plan_id}"'
)
def step_attachment_with_plan(
context: Context, val_name: str, res_id: str, plan_id: str
) -> None:
att = context.attachment_repo.attach(
validation_name=val_name,
resource_id=res_id,
mode="required",
plan_id=plan_id,
)
context.tool_session.commit()
context.last_attachment = att
# ---------------------------------------------------------------------------
# When: registration
# ---------------------------------------------------------------------------
@when("the tool is registered through the tool registry repository")
def step_register_tool(context: Context) -> None:
context.last_error = None
try:
context.tool_repo.create(context.tool_dict)
context.tool_session.commit()
except Exception as exc:
context.last_error = exc
@when('a second tool with the same registry name "{name}" is registered')
def step_register_duplicate_tool(context: Context, name: str) -> None:
dup = _make_tool_dict(name=name, source="builtin")
context.last_error = None
try:
context.tool_repo.create(dup)
context.tool_session.commit()
except Exception as exc:
context.last_error = exc
# ---------------------------------------------------------------------------
# When: retrieval
# ---------------------------------------------------------------------------
@when('the tool is looked up by registry name "{name}"')
def step_lookup_tool(context: Context, name: str) -> None:
context.returned_tool = context.tool_repo.get_by_name(name)
# ---------------------------------------------------------------------------
# When: listing
# ---------------------------------------------------------------------------
@when("all tools are listed from the tool registry")
def step_list_all_tools(context: Context) -> None:
context.returned_tools = context.tool_repo.list_all()
@when('tools in the "{ns}" namespace are listed from the registry')
def step_list_tools_ns(context: Context, ns: str) -> None:
context.returned_tools = context.tool_repo.list_all(namespace=ns)
@when('tools of type "{tool_type}" are listed from the registry')
def step_list_tools_type(context: Context, tool_type: str) -> None:
context.returned_tools = context.tool_repo.list_all(tool_type=tool_type)
@when('tools with source "{source}" are listed from the registry')
def step_list_tools_source(context: Context, source: str) -> None:
context.returned_tools = context.tool_repo.list_all(source=source)
# ---------------------------------------------------------------------------
# When: update
# ---------------------------------------------------------------------------
@when('the tool description is updated to "{desc}"')
def step_update_tool_desc(context: Context, desc: str) -> None:
context.tool_dict["description"] = desc
context.tool_repo.update(context.tool_dict)
context.tool_session.commit()
@when('the tool is updated with a new resource binding for slot "{slot_name}"')
def step_update_tool_binding(context: Context, slot_name: str) -> None:
context.tool_dict["resource_bindings"] = [_make_binding(slot_name)]
context.tool_repo.update(context.tool_dict)
context.tool_session.commit()
# ---------------------------------------------------------------------------
# When: deletion
# ---------------------------------------------------------------------------
@when('the tool "{name}" is removed from the registry')
def step_remove_tool(context: Context, name: str) -> None:
context.last_error = None
try:
context.removal_result = context.tool_repo.delete(name)
except Exception as exc:
context.last_error = exc
# ---------------------------------------------------------------------------
# When: attachments
# ---------------------------------------------------------------------------
@when('the validation "{val_name}" is attached to resource "{res_id}"')
def step_attach_validation(context: Context, val_name: str, res_id: str) -> None:
context.last_attachment = context.attachment_repo.attach(
validation_name=val_name,
resource_id=res_id,
mode="required",
)
context.tool_session.commit()
@when("the validation attachment is detached by its identifier")
def step_detach_by_id(context: Context) -> None:
att_id = context.last_attachment["attachment_id"]
context.detach_result = context.attachment_repo.detach(att_id)
context.tool_session.commit()
@when('the validation attachment "{att_id}" is detached')
def step_detach_nonexistent(context: Context, att_id: str) -> None:
context.detach_result = context.attachment_repo.detach(att_id)
@when('all attachments for resource "{res_id}" are listed')
def step_list_attachments(context: Context, res_id: str) -> None:
context.returned_attachments = context.attachment_repo.list_for_resource(
resource_id=res_id
)
@when(
'project-filtered attachments for resource "{res_id}" in project "{project}" are listed'
)
def step_list_attachments_project(context: Context, res_id: str, project: str) -> None:
context.returned_attachments = context.attachment_repo.list_for_resource(
resource_id=res_id,
project_name=project,
)
@when(
'plan-filtered attachments for resource "{res_id}" with plan "{plan_id}" are listed'
)
def step_list_attachments_plan(context: Context, res_id: str, plan_id: str) -> None:
context.returned_attachments = context.attachment_repo.list_for_resource(
resource_id=res_id,
plan_id=plan_id,
)
@when("the attachment is retrieved by its stored identifier")
def step_get_attachment_by_id(context: Context) -> None:
att_id = context.last_attachment["attachment_id"]
context.returned_attachment = context.attachment_repo.get_by_id(att_id)
# ---------------------------------------------------------------------------
# When: service layer
# ---------------------------------------------------------------------------
@given("a tool registry service backed by the repositories")
def step_tool_service(context: Context) -> None:
context.tool_service = ToolRegistryService(
tool_repo=context.tool_repo,
attachment_repo=context.attachment_repo,
)
@when("the tool is registered through the service")
def step_service_register(context: Context) -> None:
context.tool_service.register_tool(context.tool_dict)
context.tool_session.commit()
@given("the tool has been registered through the service")
def step_service_given_registered(context: Context) -> None:
context.tool_service.register_tool(context.tool_dict)
context.tool_session.commit()
@when('the tool is retrieved from the service by name "{name}"')
def step_service_get(context: Context, name: str) -> None:
context.service_returned_tool = context.tool_service.get_tool(name)
@when('the service attaches validation "{val_name}" to resource "{res_id}"')
def step_service_attach(context: Context, val_name: str, res_id: str) -> None:
context.service_attachment = context.tool_service.attach_validation(
validation_name=val_name,
resource_id=res_id,
mode="required",
)
context.tool_session.commit()
@when("the service detaches the validation attachment")
def step_service_detach(context: Context) -> None:
att_id = context.service_attachment["attachment_id"]
context.service_detach_result = context.tool_service.detach_validation(att_id)
context.tool_session.commit()
@given('the service has attached validation "{val_name}" to resource "{res_id}"')
def step_service_given_attached(context: Context, val_name: str, res_id: str) -> None:
context.service_attachment = context.tool_service.attach_validation(
validation_name=val_name,
resource_id=res_id,
mode="required",
)
context.tool_session.commit()
@when('the service lists validations for resource "{res_id}"')
def step_service_list_validations(context: Context, res_id: str) -> None:
context.service_returned_attachments = (
context.tool_service.list_validations_for_resource(
resource_id=res_id,
)
)
# ---------------------------------------------------------------------------
# Then assertions
# ---------------------------------------------------------------------------
@then("the tool registry repository should not raise an error")
def step_no_tool_error(context: Context) -> None:
assert context.last_error is None, f"Unexpected error: {context.last_error}"
@then('the persisted tool should have the registry name "{name}"')
def step_tool_name(context: Context, name: str) -> None:
tool = context.tool_repo.get_by_name(name)
assert tool is not None, f"Tool {name} not found"
assert tool["name"] == name
@then('the persisted tool should have source "{source}"')
def step_tool_source(context: Context, source: str) -> None:
name = context.tool_dict["name"]
tool = context.tool_repo.get_by_name(name)
assert tool is not None
assert tool["source"] == source
@then('the persisted tool should have tool_type "{tool_type}"')
def step_tool_type(context: Context, tool_type: str) -> None:
name = context.tool_dict["name"]
tool = context.tool_repo.get_by_name(name)
assert tool is not None
assert tool["tool_type"] == tool_type
@then("the persisted tool should have {count:d} resource binding")
def step_tool_binding_count(context: Context, count: int) -> None:
name = context.tool_dict["name"]
tool = context.tool_repo.get_by_name(name)
assert tool is not None
assert len(tool.get("resource_bindings", [])) == count
@then('a DuplicateToolError should be raised mentioning "{name}"')
def step_dup_tool_error(context: Context, name: str) -> None:
assert isinstance(context.last_error, DuplicateToolError), (
f"Expected DuplicateToolError, got {type(context.last_error)}"
)
assert name in str(context.last_error)
@then('the returned tool should have the registry name "{name}"')
def step_returned_tool_name(context: Context, name: str) -> None:
assert context.returned_tool is not None
assert context.returned_tool["name"] == name
@then("no tool should be returned from the registry")
def step_no_tool_returned(context: Context) -> None:
assert context.returned_tool is None
@then("{count:d} tools should be returned from the registry")
def step_tool_count(context: Context, count: int) -> None:
assert len(context.returned_tools) == count
@then('the returned tool names from the registry should include "{name}"')
def step_tool_names_include(context: Context, name: str) -> None:
names = [t["name"] for t in context.returned_tools]
assert name in names
@then('the persisted tool should have description "{desc}"')
def step_tool_desc(context: Context, desc: str) -> None:
name = context.tool_dict["name"]
tool = context.tool_repo.get_by_name(name)
assert tool is not None
assert tool["description"] == desc
@then('the persisted tool binding slot should be "{slot_name}"')
def step_tool_binding_slot(context: Context, slot_name: str) -> None:
name = context.tool_dict["name"]
tool = context.tool_repo.get_by_name(name)
assert tool is not None
bindings = tool.get("resource_bindings", [])
assert len(bindings) > 0
assert bindings[0]["slot_name"] == slot_name
@then("the removal should return true from the registry")
def step_removal_true(context: Context) -> None:
assert context.last_error is None
assert context.removal_result is True
@then("the removal should return false from the registry")
def step_removal_false(context: Context) -> None:
assert context.last_error is None
assert context.removal_result is False
@then("the tool should no longer exist in the registry")
def step_tool_not_found(context: Context) -> None:
name = context.tool_dict["name"]
tool = context.tool_repo.get_by_name(name)
assert tool is None
@then('a ToolInUseError should be raised mentioning "{name}"')
def step_tool_in_use_error(context: Context, name: str) -> None:
assert isinstance(context.last_error, ToolInUseError), (
f"Expected ToolInUseError, got {type(context.last_error)}"
)
assert name in str(context.last_error)
@then("the attachment should have a valid ULID identifier")
def step_attachment_ulid(context: Context) -> None:
att_id = context.last_attachment["attachment_id"]
assert len(att_id) == 26
@then('the attachment should reference resource "{res_id}"')
def step_attachment_resource(context: Context, res_id: str) -> None:
assert context.last_attachment["resource_id"] == res_id
@then("the detachment should return true")
def step_detach_true(context: Context) -> None:
assert context.detach_result is True
@then("the non-existent detachment should return false")
def step_detach_false(context: Context) -> None:
assert context.detach_result is False
@then("{count:d} attachment should be returned")
def step_attachment_count(context: Context, count: int) -> None:
assert len(context.returned_attachments) == count
@then('the retrieved attachment should reference validation "{val_name}"')
def step_retrieved_att_val(context: Context, val_name: str) -> None:
assert context.returned_attachment is not None
assert context.returned_attachment["validation_name"] == val_name
@then('the service-returned tool should have the registry name "{name}"')
def step_service_tool_name(context: Context, name: str) -> None:
assert context.service_returned_tool is not None
assert context.service_returned_tool["name"] == name
@then("the service detachment should return true")
def step_service_detach_result(context: Context) -> None:
assert context.service_detach_result is True
@then("the service should return {count:d} validation attachment")
def step_service_attachment_count(context: Context, count: int) -> None:
assert len(context.service_returned_attachments) == count
+257
View File
@@ -0,0 +1,257 @@
@phase1 @domain @repository @tool_registry
Feature: Tool Registry Persistence
As a system operator managing tools and validations
I want the tool registry to reliably persist, retrieve, and filter tools
So that the execution layer can discover and invoke tools at runtime
Background:
Given a clean in-memory database with the tool registry schema
And a tool registry repository backed by the database
And a validation attachment repository backed by the database
# ---------------------------------------------------------------------------
# Registering tools
# ---------------------------------------------------------------------------
@tool_reg_create
Scenario: A new builtin tool is registered and retrievable
Given a valid tool dict named "core/read-file" with source "builtin"
When the tool is registered through the tool registry repository
Then the tool registry repository should not raise an error
And the persisted tool should have the registry name "core/read-file"
@tool_reg_create
Scenario: A custom tool with inline code is registered
Given a valid tool dict named "local/custom-lint" with source "custom"
And the tool dict includes inline code "return {'passed': True}"
When the tool is registered through the tool registry repository
Then the persisted tool should have the registry name "local/custom-lint"
And the persisted tool should have source "custom"
@tool_reg_create
Scenario: An MCP tool is registered with server fields
Given a valid tool dict named "mcp/github-search" with source "mcp"
And the tool dict includes mcp_server "github" and mcp_tool_name "search"
When the tool is registered through the tool registry repository
Then the persisted tool should have the registry name "mcp/github-search"
@tool_reg_create
Scenario: A validation tool is registered with mode
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
When the tool is registered through the tool registry repository
Then the persisted tool should have tool_type "validation"
@tool_reg_create
Scenario: A tool is registered with resource bindings
Given a valid tool dict named "core/git-commit" with source "builtin"
And the tool dict includes a resource binding for slot "repo"
When the tool is registered through the tool registry repository
Then the persisted tool should have 1 resource binding
@tool_reg_create @error_handling
Scenario: Registering a tool with a duplicate name is rejected
Given a valid tool dict named "core/read-file" with source "builtin"
And the tool has already been registered once in the tool registry
When a second tool with the same registry name "core/read-file" is registered
Then a DuplicateToolError should be raised mentioning "core/read-file"
# ---------------------------------------------------------------------------
# Retrieving tools by name
# ---------------------------------------------------------------------------
@tool_reg_read
Scenario: A tool is retrieved by its namespaced name
Given a valid tool dict named "core/write-file" with source "builtin"
And the tool has been registered through the tool registry repository
When the tool is looked up by registry name "core/write-file"
Then the returned tool should have the registry name "core/write-file"
@tool_reg_read
Scenario: Looking up a non-existent tool name returns nothing
When the tool is looked up by registry name "core/does-not-exist"
Then no tool should be returned from the registry
# ---------------------------------------------------------------------------
# Listing tools with filters
# ---------------------------------------------------------------------------
@tool_reg_list
Scenario: All tools are listed without filters
Given the following tools have been registered:
| name | source | tool_type |
| core/read-file | builtin | tool |
| core/write-file | builtin | tool |
| local/custom-lint | custom | validation |
When all tools are listed from the tool registry
Then 3 tools should be returned from the registry
@tool_reg_list
Scenario: Tools are filtered by namespace
Given the following tools have been registered:
| name | source | tool_type |
| core/read-file | builtin | tool |
| local/custom-lint | custom | validation |
When tools in the "core" namespace are listed from the registry
Then 1 tools should be returned from the registry
And the returned tool names from the registry should include "core/read-file"
@tool_reg_list
Scenario: Tools are filtered by tool_type
Given the following tools have been registered:
| name | source | tool_type |
| core/read-file | builtin | tool |
| core/schema-check | builtin | validation |
When tools of type "validation" are listed from the registry
Then 1 tools should be returned from the registry
@tool_reg_list
Scenario: Tools are filtered by source
Given the following tools have been registered:
| name | source | tool_type |
| core/read-file | builtin | tool |
| local/custom-lint | custom | tool |
When tools with source "custom" are listed from the registry
Then 1 tools should be returned from the registry
# ---------------------------------------------------------------------------
# Updating tools
# ---------------------------------------------------------------------------
@tool_reg_update
Scenario: A tool description is updated
Given a valid tool dict named "core/read-file" with source "builtin"
And the tool has been registered through the tool registry repository
When the tool description is updated to "Updated description"
Then the persisted tool should have description "Updated description"
@tool_reg_update
Scenario: Tool resource bindings are replaced on update
Given a valid tool dict named "core/git-commit" with source "builtin"
And the tool dict includes a resource binding for slot "repo"
And the tool has been registered through the tool registry repository
When the tool is updated with a new resource binding for slot "workspace"
Then the persisted tool should have 1 resource binding
And the persisted tool binding slot should be "workspace"
# ---------------------------------------------------------------------------
# Deleting tools
# ---------------------------------------------------------------------------
@tool_reg_delete
Scenario: A tool is removed from the registry
Given a valid tool dict named "core/read-file" with source "builtin"
And the tool has been registered through the tool registry repository
When the tool "core/read-file" is removed from the registry
Then the removal should return true from the registry
And the tool should no longer exist in the registry
@tool_reg_delete
Scenario: Removing a non-existent tool returns false
When the tool "core/nonexistent" is removed from the registry
Then the removal should return false from the registry
@tool_reg_delete @error_handling
Scenario: Deleting a tool with active attachments is rejected
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the tool registry repository
And an unscoped validation attachment for "core/schema-check" on resource "res-001"
When the tool "core/schema-check" is removed from the registry
Then a ToolInUseError should be raised mentioning "core/schema-check"
# ---------------------------------------------------------------------------
# Validation attachment lifecycle
# ---------------------------------------------------------------------------
@tool_reg_attach
Scenario: A validation is attached to a resource
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the tool registry repository
When the validation "core/schema-check" is attached to resource "res-001"
Then the attachment should have a valid ULID identifier
And the attachment should reference resource "res-001"
@tool_reg_attach
Scenario: A validation is detached from a resource
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the tool registry repository
And an unscoped validation attachment for "core/schema-check" on resource "res-001"
When the validation attachment is detached by its identifier
Then the detachment should return true
@tool_reg_attach
Scenario: Detaching a non-existent attachment returns false
When the validation attachment "01ZZZZZZZZZZZZZZZZZZZZZZZZ" is detached
Then the non-existent detachment should return false
@tool_reg_attach
Scenario: Attachments are listed for a resource
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the tool registry repository
And an unscoped validation attachment for "core/schema-check" on resource "res-001"
When all attachments for resource "res-001" are listed
Then 1 attachment should be returned
@tool_reg_attach
Scenario: Attachments are scoped by project name
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the tool registry repository
And a project-scoped validation attachment for "core/schema-check" on resource "res-001" in project "my-project"
And a project-scoped validation attachment for "core/schema-check" on resource "res-001" in project "other-project"
When project-filtered attachments for resource "res-001" in project "my-project" are listed
Then 1 attachment should be returned
@tool_reg_attach
Scenario: Attachments are scoped by plan identifier
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the tool registry repository
And a plan-scoped validation attachment for "core/schema-check" on resource "res-001" in plan "01PLAN00000000000000000001"
When plan-filtered attachments for resource "res-001" with plan "01PLAN00000000000000000001" are listed
Then 1 attachment should be returned
@tool_reg_attach
Scenario: An attachment is retrieved by its identifier
Given a valid tool dict named "core/schema-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the tool registry repository
And an unscoped validation attachment for "core/schema-check" on resource "res-001"
When the attachment is retrieved by its stored identifier
Then the retrieved attachment should reference validation "core/schema-check"
# ---------------------------------------------------------------------------
# Service layer integration
# ---------------------------------------------------------------------------
@tool_reg_service
Scenario: The service registers and retrieves a tool
Given a tool registry service backed by the repositories
And a valid tool dict named "svc/echo" with source "builtin"
When the tool is registered through the service
And the tool is retrieved from the service by name "svc/echo"
Then the service-returned tool should have the registry name "svc/echo"
@tool_reg_service
Scenario: The service attaches and detaches a validation
Given a tool registry service backed by the repositories
And a valid tool dict named "svc/val-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the service
When the service attaches validation "svc/val-check" to resource "svc-res-001"
And the service detaches the validation attachment
Then the service detachment should return true
@tool_reg_service
Scenario: The service lists validations for a resource
Given a tool registry service backed by the repositories
And a valid tool dict named "svc/val-check" with source "builtin"
And the tool dict has tool_type "validation" and mode "required"
And the tool has been registered through the service
And the service has attached validation "svc/val-check" to resource "svc-res-002"
When the service lists validations for resource "svc-res-002"
Then the service should return 1 validation attachment
+222 -41
View File
@@ -883,6 +883,37 @@ The following work from the previous implementation has been completed and will
- Branch: `feature/m1-tool-builtins` (based on `feature/m1-tool-runtime-core`)
- 2026-02-14: Commit traceability sweep found no matching git log entries for skill schema/examples, skill CLI, provider actors, MCP config/adapter, actor compiler, or skill registry persistence; Section 2B traceability tasks remain open.
**2026-02-14**: Stage C1.tool.registry Complete - Tool Registry Persistence [Luis]
- Created Alembic migration `c1_001_tool_registry` (depends on `b1_001_resource_registry`) with 3 tables:
- `tools`: 22 columns, PK `name` (namespaced string e.g. `local/lint-check`), self-referential FK `wraps` -> `tools.name` (SET NULL), CHECK constraints on `tool_type` (tool/validation) and `source` (mcp/agent_skill/builtin/custom/wrapped), indexes on namespace/tool_type/source. Includes columns for inline code, MCP server/tool name, agent skill path, timeout (default 300), input/output schema JSON, capability/lifecycle JSON, transform, mode, and argument_mapping JSON.
- `tool_resource_bindings`: 10 columns, auto-increment PK, FK `tool_name` -> `tools.name` (CASCADE), columns for slot_name, resource_type, access_mode, binding_mode, static_resource, required (default true), description. Index on tool_name.
- `validation_attachments`: 8 columns, ULID PK `attachment_id`, FK `validation_name` -> `tools.name` (CASCADE), columns for resource_id, mode (required/informational CHECK), project_name, plan_id, args_json. Indexes on validation_name, resource_id, project_name.
- Added 3 ORM models to `src/cleveragents/infrastructure/database/models.py`:
- `ToolModel` (22 cols): `to_domain()` returns dict, `from_domain()` accepts dict/object with auto-namespace extraction. Relationships to `ToolResourceBindingModel` (cascade="all, delete-orphan") and `ValidationAttachmentModel`.
- `ToolResourceBindingModel` (10 cols): Nested child of ToolModel.
- `ValidationAttachmentModel` (8 cols): Nested child of ToolModel with mode CHECK constraint.
- Created 2 repository classes in `src/cleveragents/infrastructure/database/repositories.py`:
- `ToolRegistryRepository` (5 methods: create/get_by_name/list_all/update/delete): Session-factory pattern, `@database_retry` on all methods, duplicate name detection via `DuplicateToolError`, referential integrity check on delete via `ToolInUseError` (counts active validation attachments before allowing deletion). Update method replaces all child resource bindings.
- `ValidationAttachmentRepository` (4 methods: attach/detach/list_for_resource/get_by_id): ULID generation for attachment IDs, JSON serialization of args, filtering by resource_id/project_name/plan_id.
- Created `src/cleveragents/application/services/tool_registry_service.py` (192 lines):
- `ToolRegistryService` with 8 methods (5 tool CRUD + 3 validation attachment): pure dependency injection pattern, existence check on `attach_validation()` raises `NotFoundError` if validation tool not found.
- Updated `src/cleveragents/infrastructure/database/__init__.py` with 7 new exports (3 models, 2 repos, 2 errors).
- Updated `src/cleveragents/application/services/__init__.py` to export `ToolRegistryService`.
- Updated `docs/reference/database_schema.md` with 3 new table sections, updated ER diagram (tools -> bindings/attachments + self-referential wraps), updated migration chain and source locations.
- Created `features/tool_registry.feature` (27 scenarios across 7 groups: create/read/list/update/delete/attachments/service) with `features/steps/tool_registry_steps.py` (598 lines).
- Created `robot/tool_registry.robot` (4 integration tests: register-and-get, list-filter, attach-detach, duplicate-reject) with `robot/helper_tool_registry.py` (123 lines).
- Created `benchmarks/tool_registry_bench.py` (3 ASV suites, 7 benchmarks: ORM construction, CRUD operations, attachment operations).
- Key decision: `ToolModel.to_domain()` returns a dict (not a domain object) to decouple the persistence layer from the domain model until the full tool domain model integration is complete.
- Key decision: `ToolRegistryRepository.delete()` pre-checks for active `ValidationAttachmentModel` rows and raises `ToolInUseError` with count before allowing deletion, ensuring referential integrity at the application level.
- Key decision: Validation attachments are resource-scoped with optional project_name and plan_id narrowing, supporting both project-wide and plan-specific validation bindings per spec.
- Verification: lint 0 findings, typecheck 0 errors, 27 new Behave scenarios pass, 4 Robot integration tests pass.
- Branch: `feature/m3-tool-registry`
**2026-02-15**: Bugfix - Benchmark Unique Name Constraint [Luis]
- Fixed `benchmarks/resource_registry_migration_bench.py`: ASV benchmark suites that create tools/resources now use unique names per iteration (incrementing counter) to avoid UNIQUE constraint failures when ASV runs multiple iterations of the same benchmark method.
- Branch: `feature/m3-tool-registry` (appended commit)
**2026-02-17**: Bugfix - Unit Test and Benchmark Failures After Master Merge [Brent]
- **Context**: After merging master into `feature/q0-min-coverage` (commit `7ddd99b`), full `nox` run showed 2 failing sessions: `unit_tests-3.13` (1 scenario failure) and `benchmark` (multiple benchmark failures). All other sessions (lint, format, typecheck, security_scan, dead_code, integration_tests, coverage_report) passed.
@@ -916,6 +947,74 @@ The following work from the previous implementation has been completed and will
- benchmark: success (all benchmarks pass)
- **Commit**: `122af46` on `feature/q0-min-coverage``fix(tests): resolve unit test and benchmark failures`
**2026-02-17**: Task C1.schema Complete (Aditya) - Actor YAML Schema Models
- **Implementation**: Created complete actor YAML schema system in `src/cleveragents/actor/schema.py` (695 lines)
- Three actor types with clear separation: LLM (single LLM call + tools), TOOL (tool collections), GRAPH (multi-node workflows)
- Strict name validation enforcing `namespace/name` format (ADR-002 compliance)
- Comprehensive graph topology validation with cycle detection using DFS algorithm
- Type-specific field requirements enforced via Pydantic `model_validator`
- Environment variable interpolation support (`${ENV_VAR}` syntax)
- YAML I/O methods: `from_yaml_file()`, `to_yaml_file()`, `from_yaml()`
- Core models: `ActorType`, `NodeType`, `ContextView` enums; `ToolParameter`, `ToolDefinition`, `MemoryConfig`, `ContextConfigSchema`, `EdgeDefinition`, `NodeDefinition`, `RouteDefinition`, `ActorConfigSchema` Pydantic models
- **Documentation**: Comprehensive reference docs in `docs/reference/actors_schema.md` (420 lines)
- Actor type definitions, field semantics, tool node behavior
- Graph constraints: unique node IDs, entry/exit validation, cycle detection
- Validation rules with examples for valid/invalid configurations
- Error message catalog
- **Examples**: 5 YAML files in `examples/actors/`
- `simple_llm.yaml` - Minimal LLM actor (basic code review)
- `llm_with_tools.yaml` - LLM with tools, memory, context config
- `tool_collection.yaml` - TOOL actor with multiple tool references
- `simple_graph.yaml` - Basic 3-node linear graph workflow
- `graph_workflow.yaml` - Complex graph with conditionals, subgraphs, retry logic
- **Tests**: Comprehensive coverage across all test types
- **Behave**: `features/actor_schema.feature` (50+ scenarios, 300+ steps)
- Valid/invalid configurations for all three actor types
- Name validation (namespace requirement, invalid characters)
- Graph topology (unique nodes, entry/exit validation, cycle detection)
- Tool/node validation, YAML I/O round-trip, error messages
- **Robot**: `robot/actor_schema.robot` (11 test cases)
- Smoke tests for all 5 example YAMLs
- Invalid configuration rejection (missing fields, bad topology, duplicate nodes)
- **ASV**: `benchmarks/actor_schema_bench.py` (8 benchmark classes)
- Parsing/validation for minimal/full LLM, tool, graph actors
- Graph topology validation performance (cycle detection)
- File I/O operations, serialization (`model_dump`)
- **Branch**: `feature/actor-c1-schema` (11 commits across 10 sub-parts)
- **Commit messages follow Conventional Changelog format**:
1. `feat(actor): add core enums for actor schema`
2. `feat(actor): add tool and config pydantic models`
3. `feat(actor): add graph topology models with validation`
4. `feat(actor): add main actor schema and YAML I/O`
5. `docs(actor): add comprehensive actor YAML examples`
6. `docs(actor): add actor schema reference documentation`
7. `test(actor): add behave scenarios for actor schema`
8. `test(actor): add behave step definitions for actor schema`
9. `test(actor): add robot framework integration tests for actor schema`
10. `perf(actor): add asv benchmarks for actor schema`
11. `docs(actor): add c1.schema implementation notes and traceability`
- **Quality metrics**: All nox sessions pass
- lint: 0 findings (pre-existing issues in other files noted)
- typecheck: 0 errors
- security_scan: 0 findings
- unit_tests: All scenarios pass
- integration_tests: All tests pass
- benchmark: All benchmarks execute successfully
- **Implementation Notes section added** to task C1.schema in Implementation Checklist (line 2968)
- Full traceability with file paths and line numbers
- Design decisions documented with rationale
- Test coverage strategy explained
- Dependencies and open questions captured
- **Next steps**: C2.loader (Actor Registry and Loader) depends on this schema foundation
---
## Roadmap
@@ -2604,24 +2703,31 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [ ] Git [Brent]: `git branch -d feature/m3-tool-domain-robot`
- [X] Quality [Brent]: 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%.
- [ ] **COMMIT (Owner: Jeff | Group: C1.tool.registry | Branch: feature/m3-tool-registry | Planned: Day 14 | Expected: Day 20) - Commit message: "feat(tool): add tool registry persistence"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m3-tool-registry`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Add DB tables for `tools`, `tool_bindings`, and `validation_attachments` with resource-scoped attachments (`resource_id`, `validation_name`, `mode`, `created_at`) and indexes on tool name/type + resource_id.
- [ ] Code [Jeff]: Implement ToolRepository + ValidationAttachmentRepository and ToolRegistryService with attach/detach by resource and mode validation (required/informational).
- [ ] Docs [Jeff]: Update `docs/reference/database_schema.md` with tool/validation tables and constraints.
- [ ] Tests (Behave) [Jeff]: Add `features/tool_registry.feature` for register/update/remove and attachment rules.
- [ ] Tests (Robot) [Jeff]: Add `robot/tool_registry.robot` for list/show smoke tests.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/tool_registry_bench.py` for registry list performance.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [ ] Git [Jeff]: `git add .`
- [ ] Git [Jeff]: `git commit -m "feat(tool): add tool registry persistence"`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m3-tool-registry` to `master` with description "Add tool registry persistence with bindings and validation attachments.".
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git branch -d feature/m3-tool-registry`
- [ ] 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] **COMMIT (Owner: Luis | Group: C1.tool.registry | Branch: feature/m3-tool-registry | Done: Day 6, February 14, 2026 22:14:17 +0000) - Commit message: "feat(tool): add tool registry persistence"**
- [X] Git [Luis]: `git checkout master`
- [X] Git [Luis]: `git pull origin master`
- [X] Git [Luis]: `git checkout -b feature/m3-tool-registry`
- [X] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Luis]: Add DB tables for `tools`, `tool_resource_bindings`, and `validation_attachments` with resource-scoped attachments (`resource_id`, `validation_name`, `mode`, `created_at`) and indexes on tool name/type + resource_id. Migration: `alembic/versions/c1_001_tool_registry_tables.py` (depends on `b1_001_resource_registry`). 3 tables, 7 indexes, 3 CHECK constraints, 3 FKs including self-referential `wraps`.
- [X] Code [Luis]: Implement `ToolRegistryRepository` (5 methods: create/get_by_name/list_all/update/delete) + `ValidationAttachmentRepository` (4 methods: attach/detach/list_for_resource/get_by_id) and `ToolRegistryService` (8 methods) with attach/detach by resource and mode validation (required/informational). Custom errors: `DuplicateToolError`, `ToolInUseError`.
- [X] Docs [Luis]: Update `docs/reference/database_schema.md` with 3 new table sections (tools, tool_resource_bindings, validation_attachments), updated ER diagram, migration chain, and source locations.
- [X] Tests (Behave) [Luis]: Add `features/tool_registry.feature` (27 scenarios across 7 groups: create/read/list/update/delete/attachments/service) with `features/steps/tool_registry_steps.py` (598 lines).
- [X] Tests (Robot) [Luis]: Add `robot/tool_registry.robot` (4 integration tests: register-and-get, list-filter, attach-detach, duplicate-reject) with `robot/helper_tool_registry.py` (123 lines).
- [X] Tests (ASV) [Luis]: Add `benchmarks/tool_registry_bench.py` (3 ASV suites, 7 benchmarks: ORM construction, CRUD operations, attachment operations).
- [X] Quality [Luis]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
- [X] Git [Luis]: `git add .`
- [X] Git [Luis]: `git commit -m "feat(tool): add tool registry persistence"`
- [ ] Forgejo PR [Luis]: Open PR from `feature/m3-tool-registry` to `master` with description "Add tool registry persistence with bindings and validation attachments.".
- [ ] Git [Luis]: `git checkout master`
- [ ] Git [Luis]: `git branch -d feature/m3-tool-registry`
- [ ] Quality [Luis]: 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%.
**Notes (C1.tool.registry)**:
- Migration: `c1_001_tool_registry` depends on `b1_001_resource_registry`. Tables: `tools` (22 cols, PK namespaced name), `tool_resource_bindings` (10 cols, auto PK, CASCADE from tools), `validation_attachments` (8 cols, ULID PK, CASCADE from tools).
- `ToolModel.to_domain()` returns a dict (not a domain object) to decouple persistence from domain until full integration.
- `ToolRegistryRepository.delete()` pre-checks for active validation attachments and raises `ToolInUseError` with count.
- Validation attachments are resource-scoped with optional `project_name` and `plan_id` narrowing per spec.
- Second commit `fix(bench)` (2026-02-15 14:26:22 +0000) fixes `benchmarks/resource_registry_migration_bench.py` to use unique names per iteration, avoiding UNIQUE constraint failures under ASV multi-iteration runs.
- 12 files changed, +2453/-1 lines (main commit); 1 file changed, +12/-4 lines (bench fix).
- [ ] **COMMIT (Owner: Jeff | Group: C1.tool.binding | Branch: feature/m3-tool-binding | Planned: Day 15 | Expected: Day 21) - Commit message: "feat(tool): add resource binding resolution"**
- [ ] Git [Jeff]: `git checkout master`
@@ -2923,29 +3029,29 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [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%.
**Parallel Group C1: Actor Schema & Examples [Aditya + Jeff]** (start Day 5; C2 depends on this)
- [ ] **COMMIT (Owner: Aditya | Group: C1.schema | Branch: feature/m3-actor-schema-examples | Planned: Day 12 | Expected: Day 17) - Commit message: "feat(actor): add actor yaml schema models"**
- [ ] Git [Aditya]: `git checkout master`
- [ ] Git [Aditya]: `git pull origin master`
- [ ] Git [Aditya]: `git checkout -b feature/m3-actor-schema-examples`
- [ ] Git [Aditya]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Aditya]: Add schema models (ActorType, NodeType, ContextView, ToolDefinition, RouteDefinition, ActorConfigSchema) with strict validation in `src/cleveragents/actor/schema.py`.
- [ ] Code [Aditya]: Add tool-node schema fields that reference Tool Registry names, including validation nodes, and require input/output schema presence.
- [ ] Code [Aditya]: Add hierarchical graph node schema (`children`, `edges`, `entrypoint`, `exitpoints`) to support nested actors and subgraphs.
- [ ] Code [Aditya]: Add YAML load/serialize helpers and schema version guard.
- [ ] Code [Aditya]: Add graph validation rules (entrypoint exists, exitpoints reachable, no orphan nodes).
- [ ] Code [Aditya]: Add cycle detection for actor graphs and explicit errors for invalid loops.
- [ ] Docs [Aditya]: Add `docs/reference/actors_schema.md` with field definitions, tool node semantics, and graph constraints.
- [ ] Docs [Aditya]: Add graph validation examples (valid/invalid) and error messages.
- [ ] Tests (Behave) [Aditya]: Add `features/actor_schema.feature` scenarios for validation and topology errors.
- [ ] Tests (Robot) [Aditya]: Add `robot/actor_schema.robot` YAML load smoke test.
- [ ] Tests (ASV) [Aditya]: Add `benchmarks/actor_schema_bench.py` for YAML validation cost.
- [ ] Quality [Aditya]: Run `nox` (all default sessions, including benchmark).
- [ ] Git [Aditya]: `git add .` (only after coverage check passes)
- [ ] Git [Aditya]: `git commit -m "feat(actor): add actor yaml schema models"`.
- [ ] Forgejo PR [Aditya]: Open PR from `feature/m3-actor-schema-examples` to `master` with description "Add actor YAML schema models, validation rules, and tests.".
- [ ] Git [Aditya]: `git checkout master`
- [ ] Git [Aditya]: `git branch -d feature/m3-actor-schema-examples`
- [ ] Quality [Aditya]: 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] **COMMIT (Owner: Aditya | Group: C1.schema | Branch: feature/m3-actor-schema-examples | Planned: Day 12 | Expected: Day 17) - Commit message: "feat(actor): add actor yaml schema models"**
- [X] Git [Aditya]: `git checkout master`
- [X] Git [Aditya]: `git pull origin master`
- [X] Git [Aditya]: `git checkout -b feature/m3-actor-schema-examples`
- [X] Git [Aditya]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Aditya]: Add schema models (ActorType, NodeType, ContextView, ToolDefinition, RouteDefinition, ActorConfigSchema) with strict validation in `src/cleveragents/actor/schema.py`.
- [X] Code [Aditya]: Add tool-node schema fields that reference Tool Registry names, including validation nodes, and require input/output schema presence.
- [X] Code [Aditya]: Add hierarchical graph node schema (`children`, `edges`, `entrypoint`, `exitpoints`) to support nested actors and subgraphs.
- [X] Code [Aditya]: Add YAML load/serialize helpers and schema version guard.
- [X] Code [Aditya]: Add graph validation rules (entrypoint exists, exitpoints reachable, no orphan nodes).
- [X] Code [Aditya]: Add cycle detection for actor graphs and explicit errors for invalid loops.
- [X] Docs [Aditya]: Add `docs/reference/actors_schema.md` with field definitions, tool node semantics, and graph constraints.
- [X] Docs [Aditya]: Add graph validation examples (valid/invalid) and error messages.
- [X] Tests (Behave) [Aditya]: Add `features/actor_schema.feature` scenarios for validation and topology errors.
- [X] Tests (Robot) [Aditya]: Add `robot/actor_schema.robot` YAML load smoke test.
- [X] Tests (ASV) [Aditya]: Add `benchmarks/actor_schema_bench.py` for YAML validation cost.
- [X] Quality [Aditya]: Run `nox` (all default sessions, including benchmark).
- [X] Git [Aditya]: `git add .` (only after coverage check passes)
- [X] Git [Aditya]: `git commit -m "feat(actor): add actor yaml schema models"`.
- [X] Forgejo PR [Aditya]: Open PR from `feature/m3-actor-schema-examples` to `master` with description "Add actor YAML schema models, validation rules, and tests.".
- [X] Git [Aditya]: `git checkout master`
- [X] Git [Aditya]: `git branch -d feature/m3-actor-schema-examples`
- [X] Quality [Aditya]: 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%.
- [ ] **COMMIT (Owner: Aditya | Group: C1.examples | Branch: feature/m3-actor-schema-examples | Planned: Day 12 | Expected: Day 18) - Commit message: "docs(actor): add actor yaml examples"**
- [ ] Git [Aditya]: `git checkout master`
- [ ] Git [Aditya]: `git pull origin master`
@@ -2965,6 +3071,81 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- [ ] Git [Aditya]: `git branch -d feature/m3-actor-schema-examples`
- [ ] Quality [Aditya]: 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%.
#### Notes: C1.schema Actor YAML Schema Models [Aditya]
**Implementation Summary:**
- Implemented complete actor YAML schema in `src/cleveragents/actor/schema.py:1-695`
- Created 5 example YAML files in `examples/actors/` directory
- Added comprehensive documentation in `docs/reference/actors_schema.md:1-420`
- Created 50+ Behave test scenarios in `features/actor_schema.feature:1-580`
- Implemented step definitions in `features/steps/actor_schema_steps.py:1-610`
- Created Robot Framework integration tests in `robot/actor_schema.robot:1-130`
- Added helper script in `robot/helper_actor_schema.py:1-57`
- Implemented ASV benchmarks in `benchmarks/actor_schema_bench.py:1-350`
**Key Design Decisions:**
1. **Three Actor Types:** LLM, TOOL, and GRAPH actors provide clear separation of concerns
- LLM actors: Single LLM call with optional tools (simple use case)
- TOOL actors: Pure tool collections without LLM interaction (tool orchestration)
- GRAPH actors: Complex multi-node workflows with conditional routing (advanced workflows)
- Rationale: Matches ADR-010 actor architecture and supports hierarchical composition
- Location: `src/cleveragents/actor/schema.py:40-55`
2. **Strict Name Validation:** All actor names must follow `namespace/name` format
- Enforced via regex pattern in `ActorConfigSchema.validate_name()` validator
- Prevents naming conflicts and enables proper namespacing per ADR-002
- Location: `src/cleveragents/actor/schema.py:510-520`
3. **Type-Specific Field Requirements:** Pydantic model_validator enforces type-specific required fields
- LLM actors require `model` field
- TOOL actors require non-empty `tools` list
- GRAPH actors require `route` definition with nodes, edges, entry/exit points
- Location: `src/cleveragents/actor/schema.py:545-580`
4. **Graph Topology Validation:** Comprehensive validation ensures valid graph structures
- Unique node IDs checked in `RouteDefinition.validate_unique_node_ids()`
- Entry/exit node validation in `RouteDefinition.validate_entry_exit_nodes()`
- Cycle detection implemented in `RouteDefinition.detect_cycles()` using DFS
- Location: `src/cleveragents/actor/schema.py:380-480`
5. **Environment Variable Support:** Actor configs support `${ENV_VAR}` interpolation
- Enabled via `env_vars` field mapping environment variable names
- Supports both string and nested object values
- Location: `src/cleveragents/actor/schema.py:630-645`
6. **YAML I/O Methods:** Provided `from_yaml_file()` and `to_yaml_file()` class methods
- Encapsulates YAML loading/saving logic with proper error handling
- Returns validated ActorConfigSchema instances
- Location: `src/cleveragents/actor/schema.py:650-690`
**Test Coverage Strategy:**
- **Behave scenarios (50+ scenarios):** Cover all validation rules, graph topology, type-specific requirements, name validation, tool/node validation, YAML I/O, error messages
- **Robot tests (11 test cases):** Smoke tests for valid actor types, invalid configurations (missing fields, bad topology)
- **ASV benchmarks (8 benchmark classes):** Measure parsing/validation performance for minimal/full configs, graph validation, file I/O, serialization
**Example YAML Files Created:**
1. `examples/actors/simple_llm.yaml` - Minimal LLM actor with basic fields
2. `examples/actors/llm_with_tools.yaml` - LLM actor with tools and memory config
3. `examples/actors/tool_collection.yaml` - TOOL actor with multiple tool references
4. `examples/actors/simple_graph.yaml` - Basic 3-node linear graph workflow
5. `examples/actors/graph_workflow.yaml` - Complex graph with conditionals, subgraphs, retry logic
**Open Questions:**
- None at this time. All deliverables complete and validated.
**Dependencies:**
- C2.loader will depend on this schema for actor registry implementation
- C8.providers will use this schema to define built-in provider actors
- All actor-related features build on this foundation
**Traceability:**
- Schema models: `src/cleveragents/actor/schema.py:1-695`
- Examples: `examples/actors/simple_llm.yaml`, `llm_with_tools.yaml`, `tool_collection.yaml`, `simple_graph.yaml`, `graph_workflow.yaml`
- Documentation: `docs/reference/actors_schema.md:1-420`
- Behave tests: `features/actor_schema.feature:1-580`, `features/steps/actor_schema_steps.py:1-610`
- Robot tests: `robot/actor_schema.robot:1-130`, `robot/helper_actor_schema.py:1-57`
- Benchmarks: `benchmarks/actor_schema_bench.py:1-350`
**Parallel Group C2: Actor Loading & Compilation [Aditya + Jeff]** (depends on C1)
**PARALLEL SUBTRACK C2.legacy [Jeff]**: Remove v2 actor config compatibility (after C1.schema)
**SEQUENTIAL NOTE**: C2.legacy must land before C2.loader/C2.compiler to avoid dual-format support.
+119
View File
@@ -0,0 +1,119 @@
*** Settings ***
Documentation Smoke tests for Actor YAML schema validation
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_actor_schema.py
*** Test Cases ***
Validate Simple LLM Actor YAML
[Documentation] Parse the simple_llm example actor YAML and assert success
[Tags] slow
${result}= Run Process ${PYTHON} ${HELPER} validate ${CURDIR}/../examples/actors/simple_llm.yaml cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-ok
Validate LLM With Tools Actor YAML
[Documentation] Parse the llm_with_tools example actor YAML and assert success
[Tags] slow
${result}= Run Process ${PYTHON} ${HELPER} validate ${CURDIR}/../examples/actors/llm_with_tools.yaml cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-ok
Validate Tool Collection Actor YAML
[Documentation] Parse the tool_collection example actor YAML and assert success
[Tags] slow
${result}= Run Process ${PYTHON} ${HELPER} validate ${CURDIR}/../examples/actors/tool_collection.yaml cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-ok
Validate Simple Graph Actor YAML
[Documentation] Parse the simple_graph example actor YAML and assert success
[Tags] slow
${result}= Run Process ${PYTHON} ${HELPER} validate ${CURDIR}/../examples/actors/simple_graph.yaml cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-ok
Validate Complex Graph Workflow Actor YAML
[Documentation] Parse the graph_workflow example actor YAML and assert success
[Tags] slow
${result}= Run Process ${PYTHON} ${HELPER} validate ${CURDIR}/../examples/actors/graph_workflow.yaml cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-ok
Reject Actor YAML Without Namespace
[Documentation] Attempt to parse actor YAML with invalid name (no namespace)
[Tags] slow
${invalid_yaml}= Set Variable ${TEMPDIR}${/}invalid_actor_no_namespace.yaml
Create File ${invalid_yaml} name: simple_actor\ntype: llm\ndescription: Test\nmodel: gpt-4\n
${result}= Run Process ${PYTHON} ${HELPER} validate-invalid ${invalid_yaml} cwd=${WORKSPACE}
Log ${result.stdout}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-expected-fail
Should Contain ${result.stdout} namespaced
Reject LLM Actor Without Model
[Documentation] Attempt to parse LLM actor without required model field
[Tags] slow
${invalid_yaml}= Set Variable ${TEMPDIR}${/}invalid_llm_no_model.yaml
Create File ${invalid_yaml} name: test/actor\ntype: llm\ndescription: Missing model\n
${result}= Run Process ${PYTHON} ${HELPER} validate-invalid ${invalid_yaml} cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-expected-fail
Should Contain ${result.stdout} model
Reject TOOL Actor Without Tools
[Documentation] Attempt to parse TOOL actor without tools list
[Tags] slow
${invalid_yaml}= Set Variable ${TEMPDIR}${/}invalid_tool_no_tools.yaml
Create File ${invalid_yaml} name: test/tools\ntype: tool\ndescription: Missing tools\n
${result}= Run Process ${PYTHON} ${HELPER} validate-invalid ${invalid_yaml} cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-expected-fail
Should Contain ${result.stdout} tool
Reject GRAPH Actor Without Route
[Documentation] Attempt to parse GRAPH actor without route definition
[Tags] slow
${invalid_yaml}= Set Variable ${TEMPDIR}${/}invalid_graph_no_route.yaml
Create File ${invalid_yaml} name: test/workflow\ntype: graph\ndescription: Missing route\nmodel: gpt-4\n
${result}= Run Process ${PYTHON} ${HELPER} validate-invalid ${invalid_yaml} cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-expected-fail
Should Contain ${result.stdout} route
Reject Graph With Duplicate Node IDs
[Documentation] Attempt to parse graph with duplicate node IDs
[Tags] slow
${invalid_yaml}= Set Variable ${TEMPDIR}${/}invalid_graph_duplicate_nodes.yaml
${yaml_content}= Catenate SEPARATOR=\n
... name: test/duplicate
... type: graph
... description: Duplicate nodes
... model: gpt-4
... route:
... ${SPACE}${SPACE}nodes:
... ${SPACE}${SPACE}${SPACE}${SPACE}- id: node1
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}type: agent
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}name: First
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}description: First node
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}config:
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}prompt: Test
... ${SPACE}${SPACE}${SPACE}${SPACE}- id: node1
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}type: agent
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}name: Duplicate
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}description: Duplicate ID
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}config:
... ${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}${SPACE}prompt: Test
... ${SPACE}${SPACE}edges: []
... ${SPACE}${SPACE}entry_node: node1
... ${SPACE}${SPACE}exit_nodes: [node1]
Create File ${invalid_yaml} ${yaml_content}
${result}= Run Process ${PYTHON} ${HELPER} validate-invalid ${invalid_yaml} cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-schema-expected-fail
Should Contain ${result.stdout} Duplicate
+56
View File
@@ -0,0 +1,56 @@
"""Robot Framework helper for actor YAML schema validation.
Provides a CLI-style interface for Robot to invoke schema validation
on actor YAML files. Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_actor_schema.py validate <yaml_file>
python robot/helper_actor_schema.py validate-invalid <yaml_file>
"""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.actor.schema import ActorConfigSchema # noqa: E402
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 3:
print("Usage: helper_actor_schema.py <validate|validate-invalid> <file>")
return 1
command = sys.argv[1]
yaml_path = sys.argv[2]
if command == "validate":
try:
config = ActorConfigSchema.from_yaml_file(yaml_path)
print(f"actor-schema-ok: {config.name}")
return 0
except Exception as exc:
print(f"actor-schema-fail: {exc}")
return 1
if command == "validate-invalid":
try:
ActorConfigSchema.from_yaml_file(yaml_path)
print("actor-schema-unexpected-success")
return 1
except Exception as exc:
print(f"actor-schema-expected-fail: {exc}")
return 0
print(f"Unknown command: {command}")
return 1
if __name__ == "__main__":
sys.exit(main())
+122
View File
@@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""Helper script for Robot Framework tool registry smoke tests."""
from __future__ import annotations
import sys
from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
DuplicateToolError,
ToolRegistryRepository,
ValidationAttachmentRepository,
)
def _setup():
"""Create in-memory DB and return (session, tool_repo, attachment_repo)."""
engine = create_engine("sqlite:///:memory:", echo=False)
Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, expire_on_commit=False)
session = factory()
tool_repo = ToolRegistryRepository(session_factory=lambda: session)
attachment_repo = ValidationAttachmentRepository(session_factory=lambda: session)
return session, tool_repo, attachment_repo
def _make_tool(name, source="builtin", tool_type="tool", mode=None):
now = datetime.now().isoformat()
d = {
"name": name,
"description": f"Test tool {name}",
"tool_type": tool_type,
"source": source,
"timeout": 300,
"created_at": now,
"updated_at": now,
"resource_bindings": [],
}
if mode:
d["mode"] = mode
return d
def cmd_register_and_get():
session, repo, _ = _setup()
tool = _make_tool("smoke/read-file")
repo.create(tool)
session.commit()
got = repo.get_by_name("smoke/read-file")
assert got is not None
assert got["name"] == "smoke/read-file"
print("tool-registry-ok")
def cmd_list_filter():
session, repo, _ = _setup()
repo.create(_make_tool("core/tool-a"))
repo.create(_make_tool("local/tool-b", source="custom"))
session.commit()
core_tools = repo.list_all(namespace="core")
assert len(core_tools) == 1
all_tools = repo.list_all()
assert len(all_tools) == 2
print("tool-list-ok")
def cmd_attach_detach():
session, repo, att_repo = _setup()
repo.create(_make_tool("smoke/val-check", tool_type="validation", mode="required"))
session.commit()
att = att_repo.attach(
validation_name="smoke/val-check",
resource_id="res-001",
mode="required",
)
session.commit()
assert att["attachment_id"]
result = att_repo.detach(att["attachment_id"])
session.commit()
assert result is True
print("attach-detach-ok")
def cmd_duplicate_reject():
session, repo, _ = _setup()
repo.create(_make_tool("smoke/dup-test"))
session.commit()
try:
repo.create(_make_tool("smoke/dup-test"))
session.commit()
print("ERROR: should have raised")
sys.exit(1)
except DuplicateToolError:
print("duplicate-reject-ok")
def main():
if len(sys.argv) < 2:
print("Usage: helper_tool_registry.py <command>", file=sys.stderr)
sys.exit(1)
cmd = sys.argv[1]
commands = {
"register-and-get": cmd_register_and_get,
"list-filter": cmd_list_filter,
"attach-detach": cmd_attach_detach,
"duplicate-reject": cmd_duplicate_reject,
}
if cmd not in commands:
print(f"Unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
commands[cmd]()
if __name__ == "__main__":
main()
+41
View File
@@ -0,0 +1,41 @@
*** Settings ***
Documentation Smoke tests for Tool Registry persistence layer
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_tool_registry.py
*** Test Cases ***
Register And Retrieve A Tool
[Documentation] Register a builtin tool and retrieve it by name
${result}= Run Process ${PYTHON} ${HELPER} register-and-get cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tool-registry-ok
List Tools With Namespace Filter
[Documentation] Register tools and list with namespace filter
${result}= Run Process ${PYTHON} ${HELPER} list-filter cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tool-list-ok
Attach And Detach Validation
[Documentation] Attach a validation to a resource and detach it
${result}= Run Process ${PYTHON} ${HELPER} attach-detach cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} attach-detach-ok
Reject Duplicate Tool Name
[Documentation] Verify duplicate tool names are rejected
${result}= Run Process ${PYTHON} ${HELPER} duplicate-reject cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} duplicate-reject-ok
+711
View File
@@ -0,0 +1,711 @@
"""
Actor YAML Schema Definitions for CleverAgents v3.
This module defines the Pydantic models and enums that validate actor configuration
files. Actors are the v3 replacement for v2's complex reactive stream configurations,
providing a simpler, more declarative way to define AI agent behaviors.
Core Components:
Enums:
- ActorType: LLM, TOOL, or GRAPH execution models
- NodeType: AGENT, TOOL, CONDITIONAL, or SUBGRAPH node types
- ContextView: STRATEGIST, EXECUTOR, REVIEWER, or FULL context filtering
Tool Models:
- ToolParameter: Parameter definitions for inline tools
- ToolDefinition: Complete inline tool with Python code
Graph Models:
- EdgeDefinition: Connections between nodes with conditional routing
- NodeDefinition: Node specifications (agent, tool, conditional, subgraph)
- RouteDefinition: Complete graph topology with validation
Configuration:
- MemoryConfig: Conversation history settings
- ContextConfigSchema: File inclusion and context window configuration
- ActorConfigSchema: Main schema bringing all components together
Usage:
>>> from cleveragents.actor.schema import ActorConfigSchema
>>> config = ActorConfigSchema.from_yaml_file("path/to/actor.yaml")
>>> print(config.name, config.type)
Author: CleverAgents Team
Version: 3.0.0
"""
from __future__ import annotations
from enum import StrEnum
from pathlib import Path
from typing import Any
import yaml
from pydantic import BaseModel, Field, field_validator, model_validator
class ActorType(StrEnum):
"""
Type of actor determining execution behavior.
Actors in v3 are compiled to LangGraph workflows. The type determines
how the actor is compiled and executed:
- LLM: Single LLM call with optional tools (simplest)
- TOOL: Collection of callable tools with agent orchestration
- GRAPH: Multi-node StateGraph with conditional routing (most flexible)
Examples:
>>> actor_config = {"type": ActorType.LLM, "model": "gpt-4"}
>>> if actor_config["type"] == ActorType.GRAPH:
... # Compile to multi-node graph
"""
LLM = "llm" # Single LLM agent with system prompt and optional tools
TOOL = "tool" # Collection of callable tools (file ops, API calls, etc.)
GRAPH = "graph" # Multi-node StateGraph with routing and conditionals
class NodeType(StrEnum):
"""
Type of node in a graph actor.
Graph actors (ActorType.GRAPH) are composed of multiple nodes connected
by edges. Each node performs a specific operation in the workflow:
- AGENT: Invokes an LLM to make decisions or generate content
- TOOL: Executes a specific tool (file operations, API calls, etc.)
- CONDITIONAL: Routes execution based on conditions (if/else logic)
- SUBGRAPH: Embeds another actor as a nested workflow (composition)
Node types enable complex workflows like:
code_writer (AGENT) run_tests (TOOL) check_passed (CONDITIONAL)
if passed: style_check (SUBGRAPH), else: back to code_writer
Examples:
>>> node_def = {"type": NodeType.AGENT, "prompt": "Write Python code"}
>>> if node_def["type"] == NodeType.SUBGRAPH:
... # Load and compile referenced actor
"""
AGENT = "agent" # LLM agent node (makes AI-powered decisions)
TOOL = "tool" # Tool execution node (runs specific function)
CONDITIONAL = "conditional" # Routing node (evaluates conditions)
SUBGRAPH = "subgraph" # Nested actor reference (hierarchical composition)
class ContextView(StrEnum):
"""
Role-based context filtering for actors.
Different actor roles need different levels of context. Providing too much
context wastes tokens and degrades performance; too little causes errors.
ContextView allows actors to request only the information they need.
- STRATEGIST: High-level planning view (project structure, goals, constraints)
- EXECUTOR: Implementation view (code, files, specific tasks)
- REVIEWER: Validation view (changes, diffs, test results)
- FULL: Complete view (all available context, use sparingly)
Examples:
>>> strategist = {"context_view": ContextView.STRATEGIST}
>>> executor = {"context_view": ContextView.EXECUTOR}
>>> # Strategist sees project-level info, executor sees file-level details
"""
STRATEGIST = "strategist" # Planning and strategy context
EXECUTOR = "executor" # Implementation and execution context
REVIEWER = "reviewer" # Review and validation context
FULL = "full" # Complete context (use sparingly)
# ============================================================================
# Tool Models (for inline tool definitions in actors)
# ============================================================================
class ToolParameter(BaseModel):
"""
Parameter definition for inline tool functions.
Used in ToolDefinition to specify input parameters for Python code tools.
Supports type hints and default values for tool inputs.
Attributes:
name: Parameter name (must be valid Python identifier)
type: Python type annotation as string (e.g., "str", "int", "list[str]")
description: Human-readable parameter description
required: Whether parameter must be provided (default: True)
default: Default value if not provided (only for optional params)
Examples:
>>> param = ToolParameter(
... name="input_file",
... type="str",
... description="Path to input file",
... required=True
... )
"""
name: str = Field(..., description="Parameter name")
type: str = Field(..., description="Python type annotation")
description: str = Field(..., description="Parameter description")
required: bool = Field(default=True, description="Whether required")
default: Any | None = Field(default=None, description="Default value")
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
"""Ensure parameter name is a valid Python identifier."""
if not v.isidentifier():
msg = f"Parameter name must be valid Python identifier: {v}"
raise ValueError(msg)
return v
class ToolDefinition(BaseModel):
"""
Inline tool definition with Python code.
Allows defining simple tools directly in actor YAML files without
creating separate tool modules. Useful for actor-specific utilities.
Attributes:
name: Tool name (namespaced format: "namespace/tool_name")
description: What the tool does (used in LLM tool selection)
parameters: List of input parameters
code: Python code implementing the tool (must define a function)
Examples:
>>> tool = ToolDefinition(
... name="utils/count_lines",
... description="Count lines in a file",
... parameters=[
... ToolParameter(name="file_path", type="str", description="File")
... ],
... code="def count_lines(file_path: str) -> int:\\n ..."
... )
"""
name: str = Field(..., description="Tool name (namespaced)")
description: str = Field(..., description="Tool description")
parameters: list[ToolParameter] = Field(
default_factory=list, description="Tool parameters"
)
code: str = Field(..., description="Python code for tool")
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
"""Ensure tool name follows namespace/name format."""
if "/" not in v:
msg = f"Tool name must be namespaced (namespace/name): {v}"
raise ValueError(msg)
return v
# ============================================================================
# Configuration Models (memory and context settings)
# ============================================================================
class MemoryConfig(BaseModel):
"""
Conversation history and memory settings for actors.
Controls how much conversation history is retained and passed to the LLM.
Balances context quality with token usage.
Attributes:
enabled: Whether to maintain conversation history (default: True)
max_messages: Maximum messages to retain (None = unlimited)
max_tokens: Maximum tokens in history (None = unlimited)
summarize_old: Whether to summarize old messages (default: False)
Examples:
>>> memory = MemoryConfig(
... enabled=True,
... max_messages=50,
... max_tokens=4000
... )
"""
enabled: bool = Field(default=True, description="Enable conversation memory")
max_messages: int | None = Field(default=None, description="Max messages to retain")
max_tokens: int | None = Field(default=None, description="Max tokens in history")
summarize_old: bool = Field(default=False, description="Summarize old messages")
class ContextConfigSchema(BaseModel):
"""
File inclusion and context window configuration.
Defines which files/directories to include in actor context and how
to manage the context window size.
Attributes:
include_files: List of file paths to include in context
include_dirs: List of directory paths to include in context
exclude_patterns: Glob patterns to exclude from context
max_context_tokens: Maximum context window size (None = model default)
Examples:
>>> context = ContextConfigSchema(
... include_files=["README.md", "src/main.py"],
... include_dirs=["src/", "tests/"],
... exclude_patterns=["**/__pycache__/**", "*.pyc"],
... max_context_tokens=8000
... )
"""
include_files: list[str] = Field(
default_factory=list, description="Files to include"
)
include_dirs: list[str] = Field(
default_factory=list, description="Directories to include"
)
exclude_patterns: list[str] = Field(
default_factory=list, description="Exclusion patterns"
)
max_context_tokens: int | None = Field(
default=None, description="Max context tokens"
)
# ============================================================================
# Graph Models (for ActorType.GRAPH - multi-node workflows)
# ============================================================================
class EdgeDefinition(BaseModel):
"""
Edge connection between nodes in a graph actor.
Defines transitions between nodes with optional conditional routing.
Edges determine the flow of execution through the graph.
Attributes:
from_node: Source node ID
to_node: Target node ID
condition: Optional Python expression for conditional routing
priority: Edge priority for multiple outgoing edges (higher = first)
Examples:
>>> # Simple unconditional edge
>>> edge = EdgeDefinition(from_node="agent1", to_node="tool1")
>>>
>>> # Conditional edge with routing logic
>>> edge = EdgeDefinition(
... from_node="checker",
... to_node="fix_errors",
... condition="state.get('has_errors') == True",
... priority=10
... )
"""
from_node: str = Field(..., description="Source node ID")
to_node: str = Field(..., description="Target node ID")
condition: str | None = Field(
default=None, description="Conditional routing expression"
)
priority: int = Field(default=0, description="Edge priority")
class NodeDefinition(BaseModel):
"""
Node definition in a graph actor.
Represents a single step in a multi-node workflow. Nodes can be
agents (LLM calls), tools (function execution), conditionals (routing),
or subgraphs (nested actors).
Attributes:
id: Unique node identifier within the graph
type: Node type (AGENT, TOOL, CONDITIONAL, SUBGRAPH)
name: Human-readable node name
description: Node purpose and behavior
config: Type-specific configuration (depends on node type)
Node Type Configurations:
AGENT: {"model": "gpt-4", "prompt": "...", "tools": [...]}
TOOL: {"tool_name": "namespace/tool", "parameters": {...}}
CONDITIONAL: {"conditions": [{"check": "...", "route_to": "..."}]}
SUBGRAPH: {"actor_path": "path/to/actor.yaml"}
Examples:
>>> # Agent node
>>> agent = NodeDefinition(
... id="code_writer",
... type=NodeType.AGENT,
... name="Code Writer",
... description="Writes Python code",
... config={"model": "gpt-4", "prompt": "Write clean Python code"}
... )
>>>
>>> # Tool node
>>> tool = NodeDefinition(
... id="run_tests",
... type=NodeType.TOOL,
... name="Test Runner",
... description="Executes pytest",
... config={"tool_name": "testing/run_pytest"}
... )
"""
id: str = Field(..., description="Node ID")
type: NodeType = Field(..., description="Node type")
name: str = Field(..., description="Node name")
description: str = Field(..., description="Node description")
config: dict[str, Any] = Field(default_factory=dict, description="Node config")
@field_validator("id")
@classmethod
def validate_id(cls, v: str) -> str:
"""Ensure node ID is a valid identifier."""
if not v.replace("_", "").replace("-", "").isalnum():
msg = f"Node ID must be alphanumeric with underscores/hyphens: {v}"
raise ValueError(msg)
return v
class RouteDefinition(BaseModel):
"""
Complete graph topology for ActorType.GRAPH actors.
Defines the full multi-node workflow including all nodes, edges,
and entry/exit points. Includes validation for cycle detection
and reachability.
Attributes:
nodes: List of all nodes in the graph
edges: List of all edges connecting nodes
entry_node: ID of the starting node
exit_nodes: List of IDs for terminal nodes
Validation:
- All nodes must have unique IDs
- Entry node must exist in nodes
- All exit nodes must exist in nodes
- All edge references must point to valid nodes
- Graph must be acyclic (no cycles allowed)
- All nodes must be reachable from entry_node
Examples:
>>> route = RouteDefinition(
... nodes=[
... NodeDefinition(
... id="start",
... type=NodeType.AGENT,
... name="Planner",
... description="Plans tasks",
... config={"model": "gpt-4"}
... ),
... NodeDefinition(
... id="execute",
... type=NodeType.TOOL,
... name="Executor",
... description="Runs tasks",
... config={"tool_name": "exec/run"}
... )
... ],
... edges=[
... EdgeDefinition(from_node="start", to_node="execute")
... ],
... entry_node="start",
... exit_nodes=["execute"]
... )
"""
nodes: list[NodeDefinition] = Field(..., description="Graph nodes")
edges: list[EdgeDefinition] = Field(..., description="Graph edges")
entry_node: str = Field(..., description="Entry node ID")
exit_nodes: list[str] = Field(..., description="Exit node IDs")
@field_validator("nodes")
@classmethod
def validate_unique_ids(cls, v: list[NodeDefinition]) -> list[NodeDefinition]:
"""Ensure all node IDs are unique."""
ids = [node.id for node in v]
if len(ids) != len(set(ids)):
duplicates = [id for id in ids if ids.count(id) > 1]
msg = f"Duplicate node IDs found: {set(duplicates)}"
raise ValueError(msg)
return v
def validate_references(self) -> None:
"""
Validate all node references in edges and entry/exit points.
Raises:
ValueError: If any reference points to non-existent node
"""
node_ids = {node.id for node in self.nodes}
# Validate entry node
if self.entry_node not in node_ids:
msg = f"Entry node '{self.entry_node}' not found in nodes"
raise ValueError(msg)
# Validate exit nodes
for exit_node in self.exit_nodes:
if exit_node not in node_ids:
msg = f"Exit node '{exit_node}' not found in nodes"
raise ValueError(msg)
# Validate edge references
for edge in self.edges:
if edge.from_node not in node_ids:
msg = f"Edge from_node '{edge.from_node}' not found in nodes"
raise ValueError(msg)
if edge.to_node not in node_ids:
msg = f"Edge to_node '{edge.to_node}' not found in nodes"
raise ValueError(msg)
def detect_cycles(self) -> list[str]:
"""
Detect cycles in the graph using DFS.
Returns:
List of node IDs involved in cycles (empty if acyclic)
"""
# Build adjacency list
graph: dict[str, list[str]] = {node.id: [] for node in self.nodes}
for edge in self.edges:
graph[edge.from_node].append(edge.to_node)
# DFS cycle detection
visited: set[str] = set()
rec_stack: set[str] = set()
cycle_nodes: list[str] = []
def dfs(node: str) -> bool:
visited.add(node)
rec_stack.add(node)
for neighbor in graph.get(node, []):
if neighbor not in visited:
if dfs(neighbor):
return True
elif neighbor in rec_stack:
cycle_nodes.append(neighbor)
return True
rec_stack.remove(node)
return False
for node in graph:
if node not in visited and dfs(node):
break
return cycle_nodes
# ============================================================================
# Main Actor Configuration Schema
# ============================================================================
class ActorConfigSchema(BaseModel):
"""
Main actor configuration schema bringing all components together.
This is the top-level model for actor YAML files. Actors are the v3
replacement for v2's reactive streams, providing a simpler declarative
way to define AI agent behaviors.
Attributes:
name: Actor name (namespaced format: "namespace/actor_name")
type: Actor type (LLM, TOOL, or GRAPH)
description: What the actor does
version: Schema version (default: "1.0")
model: LLM model name (required for LLM/GRAPH types)
system_prompt: System prompt for LLM actors
tools: List of tool references or inline definitions
context_view: Role-based context filtering
memory: Memory and conversation history settings
context: File inclusion and context window settings
route: Graph topology (required for GRAPH type)
env_vars: Environment variable mappings
Type-Specific Requirements:
ActorType.LLM: Requires model, optional system_prompt and tools
ActorType.TOOL: Requires tools list
ActorType.GRAPH: Requires model and route
Examples:
>>> # Simple LLM actor
>>> actor = ActorConfigSchema(
... name="assistants/code_reviewer",
... type=ActorType.LLM,
... description="Reviews Python code for best practices",
... model="gpt-4",
... system_prompt="You are an expert Python code reviewer"
... )
>>>
>>> # Graph actor with multiple nodes
>>> actor = ActorConfigSchema(
... name="workflows/test_driven_dev",
... type=ActorType.GRAPH,
... description="Test-driven development workflow",
... model="gpt-4",
... route=RouteDefinition(...)
... )
"""
name: str = Field(..., description="Actor name (namespaced)")
type: ActorType = Field(..., description="Actor type")
description: str = Field(..., description="Actor description")
version: str = Field(default="1.0", description="Schema version")
# LLM configuration
model: str | None = Field(default=None, description="LLM model name")
system_prompt: str | None = Field(default=None, description="System prompt")
# Tool configuration
tools: list[str | ToolDefinition] = Field(
default_factory=list, description="Tool references or definitions"
)
# Context and memory
context_view: ContextView | None = Field(
default=None, description="Context filtering view"
)
memory: MemoryConfig = Field(
default_factory=MemoryConfig, description="Memory settings"
)
context: ContextConfigSchema = Field(
default_factory=ContextConfigSchema, description="Context settings"
)
# Graph configuration
route: RouteDefinition | None = Field(
default=None, description="Graph topology (for GRAPH type)"
)
# Environment variables
env_vars: dict[str, str] = Field(
default_factory=dict, description="Environment variable mappings"
)
@field_validator("name")
@classmethod
def validate_name(cls, v: str) -> str:
"""Ensure actor name follows namespace/name format."""
if "/" not in v:
msg = f"Actor name must be namespaced (namespace/name): {v}"
raise ValueError(msg)
# Check for exactly one slash
parts = v.split("/")
if len(parts) != 2:
msg = (
f"Actor name must be namespaced with exactly one slash "
f"(namespace/name): {v}"
)
raise ValueError(msg)
namespace, name = parts
if not namespace or not name:
msg = (
f"Actor name must be namespaced with non-empty namespace and name: {v}"
)
raise ValueError(msg)
return v
@model_validator(mode="after")
def validate_type_requirements(self) -> ActorConfigSchema:
"""Validate type-specific requirements."""
# LLM actors require model
if self.type == ActorType.LLM and not self.model:
msg = "LLM actors require 'model' field"
raise ValueError(msg)
# TOOL actors require tools
if self.type == ActorType.TOOL and not self.tools:
msg = "TOOL actors require at least one tool"
raise ValueError(msg)
# GRAPH actors require model and route
if self.type == ActorType.GRAPH:
if not self.model:
msg = "GRAPH actors require 'model' field"
raise ValueError(msg)
if not self.route:
msg = "GRAPH actors require 'route' field"
raise ValueError(msg)
# Validate route references and cycles
self.route.validate_references()
cycles = self.route.detect_cycles()
if cycles:
msg = f"Graph contains cycles involving nodes: {cycles}"
raise ValueError(msg)
return self
@classmethod
def from_yaml_file(cls, file_path: str | Path) -> ActorConfigSchema:
"""
Load actor configuration from YAML file.
Args:
file_path: Path to YAML file
Returns:
Parsed and validated ActorConfigSchema
Raises:
FileNotFoundError: If file doesn't exist
yaml.YAMLError: If YAML is invalid
ValidationError: If schema validation fails
Examples:
>>> actor = ActorConfigSchema.from_yaml_file("actors/reviewer.yaml")
>>> print(actor.name, actor.type)
"""
path = Path(file_path)
if not path.exists():
msg = f"Actor file not found: {file_path}"
raise FileNotFoundError(msg)
with path.open("r", encoding="utf-8") as f:
data = yaml.safe_load(f)
return cls.model_validate(data)
def to_yaml_file(self, file_path: str | Path) -> None:
"""
Save actor configuration to YAML file.
Args:
file_path: Path to save YAML file
Examples:
>>> actor.to_yaml_file("actors/new_actor.yaml")
"""
path = Path(file_path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("w", encoding="utf-8") as f:
yaml.safe_dump(
self.model_dump(mode="json", exclude_none=True),
f,
default_flow_style=False,
sort_keys=False,
)
__all__ = [
"ActorConfigSchema",
"ActorType",
"ContextConfigSchema",
"ContextView",
"EdgeDefinition",
"MemoryConfig",
"NodeDefinition",
"NodeType",
"RouteDefinition",
"ToolDefinition",
"ToolParameter",
]
@@ -3,4 +3,10 @@
Contains service classes that orchestrate business operations.
"""
__all__ = []
from cleveragents.application.services.tool_registry_service import (
ToolRegistryService,
)
__all__ = [
"ToolRegistryService",
]
@@ -0,0 +1,192 @@
"""Service layer for managing tool and validation registrations.
Orchestrates persistence through ``ToolRegistryRepository`` and
``ValidationAttachmentRepository``, providing a clean API for CLI and
higher-level orchestrators.
"""
from __future__ import annotations
from typing import Any
from cleveragents.core.exceptions import (
NotFoundError,
)
from cleveragents.infrastructure.database.repositories import (
ToolRegistryRepository,
ValidationAttachmentRepository,
)
class ToolRegistryService:
"""Coordinate tool/validation registration and attachment lifecycle.
Delegates all persistence to the repository layer. Designed to be
injected with a ``ToolRegistryRepository`` and a
``ValidationAttachmentRepository`` (both using session-factory pattern).
"""
def __init__(
self,
tool_repo: ToolRegistryRepository,
attachment_repo: ValidationAttachmentRepository,
) -> None:
self._tool_repo = tool_repo
self._attachment_repo = attachment_repo
# -- Tool CRUD ---------------------------------------------------------
def register_tool(self, tool: Any) -> Any:
"""Register a new tool or validation.
Args:
tool: A tool dict or domain object.
Returns:
The registered tool.
Raises:
DuplicateToolError: If the name already exists.
DatabaseError: On persistence failure.
"""
return self._tool_repo.create(tool)
def update_tool(self, tool: Any) -> Any:
"""Update an existing tool definition.
Args:
tool: A tool dict or domain object with updated fields.
Returns:
The updated tool.
Raises:
DatabaseError: If the tool is not found or persistence fails.
"""
return self._tool_repo.update(tool)
def remove_tool(self, name: str) -> bool:
"""Remove a tool from the registry.
Args:
name: The namespaced name of the tool.
Returns:
``True`` if the tool was removed, ``False`` if not found.
Raises:
ToolInUseError: If validation attachments still reference
this tool.
"""
return self._tool_repo.delete(name)
def list_tools(
self,
namespace: str | None = None,
tool_type: str | None = None,
source: str | None = None,
) -> list[Any]:
"""List tools with optional filters.
Args:
namespace: Optional namespace filter.
tool_type: Optional ``tool`` or ``validation`` filter.
source: Optional source filter.
Returns:
List of tool dicts.
"""
return self._tool_repo.list_all(
namespace=namespace,
tool_type=tool_type,
source=source,
)
def get_tool(self, name: str) -> Any | None:
"""Retrieve a single tool by name.
Args:
name: The namespaced name.
Returns:
The tool dict, or ``None`` if not found.
"""
return self._tool_repo.get_by_name(name)
# -- Validation attachment lifecycle -----------------------------------
def attach_validation(
self,
validation_name: str,
resource_id: str,
mode: str = "required",
project_name: str | None = None,
plan_id: str | None = None,
args: dict[str, Any] | None = None,
) -> Any:
"""Attach a validation to a resource.
Args:
validation_name: Name of the validation tool to attach.
resource_id: Resource reference string.
mode: ``required`` or ``informational``.
project_name: Optional project scope.
plan_id: Optional plan scope.
args: Optional invocation arguments.
Returns:
The attachment dict with generated ``attachment_id``.
Raises:
NotFoundError: If the validation tool does not exist.
DatabaseError: On persistence failure.
"""
# Verify validation exists
existing = self._tool_repo.get_by_name(validation_name)
if existing is None:
raise NotFoundError(
resource_type="validation",
resource_id=validation_name,
)
return self._attachment_repo.attach(
validation_name=validation_name,
resource_id=resource_id,
mode=mode,
project_name=project_name,
plan_id=plan_id,
args=args,
)
def detach_validation(self, attachment_id: str) -> bool:
"""Detach a validation attachment.
Args:
attachment_id: The ULID of the attachment.
Returns:
``True`` if the attachment was removed.
"""
return self._attachment_repo.detach(attachment_id)
def list_validations_for_resource(
self,
resource_id: str,
project_name: str | None = None,
plan_id: str | None = None,
) -> list[Any]:
"""List validation attachments for a resource.
Args:
resource_id: The resource reference.
project_name: Optional project scope filter.
plan_id: Optional plan scope filter.
Returns:
List of attachment dicts.
"""
return self._attachment_repo.list_for_resource(
resource_id=resource_id,
project_name=project_name,
plan_id=plan_id,
)
@@ -21,6 +21,9 @@ from .models import (
ResourceEdgeModel,
ResourceModel,
ResourceTypeModel,
ToolModel,
ToolResourceBindingModel,
ValidationAttachmentModel,
get_session,
init_database,
)
@@ -34,6 +37,7 @@ from .repositories import (
DuplicatePlanError,
DuplicateResourceError,
DuplicateResourceTypeError,
DuplicateToolError,
LifecyclePlanRepository,
NamespacedProjectRepository,
PlanNotFoundError,
@@ -47,6 +51,9 @@ from .repositories import (
ResourceTypeHasResourcesError,
ResourceTypeNotFoundError,
ResourceTypeRepository,
ToolInUseError,
ToolRegistryRepository,
ValidationAttachmentRepository,
)
from .unit_of_work import UnitOfWork, UnitOfWorkContext
@@ -65,6 +72,7 @@ __all__ = [
"DuplicatePlanError",
"DuplicateResourceError",
"DuplicateResourceTypeError",
"DuplicateToolError",
"LifecycleActionModel",
"LifecyclePlanModel",
"LifecyclePlanRepository",
@@ -90,8 +98,14 @@ __all__ = [
"ResourceTypeModel",
"ResourceTypeNotFoundError",
"ResourceTypeRepository",
"ToolInUseError",
"ToolModel",
"ToolRegistryRepository",
"ToolResourceBindingModel",
"UnitOfWork",
"UnitOfWorkContext",
"ValidationAttachmentModel",
"ValidationAttachmentRepository",
"get_session",
"init_database",
]
@@ -1463,6 +1463,293 @@ class ResourceEdgeModel(Base): # type: ignore[misc]
)
# ---------------------------------------------------------------------------
# Tool Registry Models (Stage C1 - migration c1_001)
# ---------------------------------------------------------------------------
class ToolModel(Base): # type: ignore[misc]
"""Database model for tool and validation definitions.
Tools are the atomic unit of execution in CleverAgents. They share
a single naming namespace with Validations (which are just tools
with pass/fail semantics). The ``tool_type`` discriminator
distinguishes them for filtering.
Table: ``tools``
"""
__allow_unmapped__ = True
__tablename__ = "tools"
# PK: namespaced name (e.g. "local/lint-check")
name = Column(String(255), primary_key=True)
namespace = Column(String(100), nullable=False)
short_name = Column(String(150), nullable=False)
# Core fields
description = Column(Text, nullable=False)
tool_type = Column(String(20), nullable=False)
source = Column(String(20), nullable=False)
# JSON Schema for inputs/outputs
input_schema_json = Column(Text, nullable=True)
output_schema_json = Column(Text, nullable=True)
# Capability and lifecycle metadata (JSON blobs)
capability_json = Column(Text, nullable=True)
lifecycle_json = Column(Text, nullable=True)
# Source-specific fields
code = Column(Text, nullable=True)
mcp_server = Column(String(255), nullable=True)
mcp_tool_name = Column(String(255), nullable=True)
agent_skill_path = Column(String(1024), nullable=True)
# Execution
timeout = Column(Integer, nullable=False, default=300)
# Validation-specific fields
wraps = Column(
String(255),
ForeignKey("tools.name", ondelete="SET NULL"),
nullable=True,
)
transform = Column(Text, nullable=True)
mode = Column(String(20), nullable=True)
argument_mapping_json = Column(Text, nullable=True)
# Timestamps (ISO-8601 strings)
created_at = Column(String(30), nullable=False)
updated_at = Column(String(30), nullable=False)
# Relationships
resource_bindings_rel = relationship(
"ToolResourceBindingModel",
back_populates="tool",
cascade="all, delete-orphan",
)
validation_attachments_rel = relationship(
"ValidationAttachmentModel",
back_populates="tool",
cascade="all, delete-orphan",
)
__table_args__ = (
CheckConstraint(
"tool_type IN ('tool', 'validation')",
name="ck_tools_type",
),
CheckConstraint(
"source IN ('mcp', 'agent_skill', 'builtin', 'custom', 'wrapped')",
name="ck_tools_source",
),
Index("ix_tools_namespace", "namespace"),
Index("ix_tools_tool_type", "tool_type"),
Index("ix_tools_source", "source"),
)
# -- Domain conversion helpers ------------------------------------------
def to_domain(self) -> Any:
"""Convert to domain model (stub returning dict).
Returns:
A dictionary representation suitable for domain reconstruction.
"""
result: dict[str, Any] = {
"name": cast(str, self.name),
"namespace": cast(str, self.namespace),
"short_name": cast(str, self.short_name),
"description": cast(str, self.description),
"tool_type": cast(str, self.tool_type),
"source": cast(str, self.source),
"input_schema_json": cast("str | None", self.input_schema_json),
"output_schema_json": cast("str | None", self.output_schema_json),
"capability_json": cast("str | None", self.capability_json),
"lifecycle_json": cast("str | None", self.lifecycle_json),
"code": cast("str | None", self.code),
"mcp_server": cast("str | None", self.mcp_server),
"mcp_tool_name": cast("str | None", self.mcp_tool_name),
"agent_skill_path": cast("str | None", self.agent_skill_path),
"timeout": cast(int, self.timeout),
"wraps": cast("str | None", self.wraps),
"transform": cast("str | None", self.transform),
"mode": cast("str | None", self.mode),
"argument_mapping_json": cast("str | None", self.argument_mapping_json),
"created_at": cast(str, self.created_at),
"updated_at": cast(str, self.updated_at),
}
# Include resource bindings
bindings: list[dict[str, Any]] = []
for binding in self.resource_bindings_rel or []:
bindings.append(
{
"id": cast(int, binding.id),
"slot_name": cast(str, binding.slot_name),
"resource_type": cast(str, binding.resource_type),
"access_mode": cast(str, binding.access_mode),
"binding_mode": cast(str, binding.binding_mode),
"static_resource": cast("str | None", binding.static_resource),
"required": cast(bool, binding.required),
"description": cast("str | None", binding.description),
}
)
result["resource_bindings"] = bindings
return result
@classmethod
def from_domain(cls, tool: Any) -> ToolModel:
"""Create from a domain tool dict or object.
Args:
tool: A dict-like or domain object with tool fields.
Returns:
A ``ToolModel`` ready for persistence.
"""
# Support both dict and object access
def _get(obj: Any, key: str, default: Any = None) -> Any:
if isinstance(obj, dict):
return obj.get(key, default)
return getattr(obj, key, default)
now_iso = datetime.now().isoformat()
name_str: str = _get(tool, "name", "")
model = cls(
name=name_str,
namespace=name_str.split("/", 1)[0] if "/" in name_str else "",
short_name=name_str.split("/", 1)[1] if "/" in name_str else name_str,
description=_get(tool, "description", ""),
tool_type=_get(tool, "tool_type", "tool"),
source=_get(tool, "source", "builtin"),
input_schema_json=_get(tool, "input_schema_json"),
output_schema_json=_get(tool, "output_schema_json"),
capability_json=_get(tool, "capability_json"),
lifecycle_json=_get(tool, "lifecycle_json"),
code=_get(tool, "code"),
mcp_server=_get(tool, "mcp_server"),
mcp_tool_name=_get(tool, "mcp_tool_name"),
agent_skill_path=_get(tool, "agent_skill_path"),
timeout=_get(tool, "timeout", 300),
wraps=_get(tool, "wraps"),
transform=_get(tool, "transform"),
mode=_get(tool, "mode"),
argument_mapping_json=_get(tool, "argument_mapping_json"),
created_at=_get(tool, "created_at", now_iso),
updated_at=_get(tool, "updated_at", now_iso),
)
# Populate resource bindings from domain
for binding_data in _get(tool, "resource_bindings", []) or []:
model.resource_bindings_rel.append(
ToolResourceBindingModel(
slot_name=_get(binding_data, "slot_name", ""),
resource_type=_get(binding_data, "resource_type", ""),
access_mode=_get(binding_data, "access_mode", "read_only"),
binding_mode=_get(binding_data, "binding_mode", "contextual"),
static_resource=_get(binding_data, "static_resource"),
required=_get(binding_data, "required", True),
description=_get(binding_data, "description"),
created_at=now_iso,
)
)
return model
class ToolResourceBindingModel(Base): # type: ignore[misc]
"""Database model for tool resource bindings (child of tools).
Stores the typed resource slot declarations for each tool.
Table: ``tool_resource_bindings``
"""
__allow_unmapped__ = True
__tablename__ = "tool_resource_bindings"
id = Column(Integer, primary_key=True, autoincrement=True)
tool_name = Column(
String(255),
ForeignKey("tools.name", ondelete="CASCADE"),
nullable=False,
)
slot_name = Column(String(100), nullable=False)
resource_type = Column(String(255), nullable=False)
access_mode = Column(String(20), nullable=False)
binding_mode = Column(String(20), nullable=False)
static_resource = Column(String(255), nullable=True)
required = Column(Boolean, nullable=False, default=True)
description = Column(Text, nullable=True)
created_at = Column(String(30), nullable=False)
# Relationships
tool = relationship(
"ToolModel",
back_populates="resource_bindings_rel",
)
__table_args__ = (Index("ix_tool_resource_bindings_tool_name", "tool_name"),)
class ValidationAttachmentModel(Base): # type: ignore[misc]
"""Database model for validation attachments.
Runtime attachments linking validations to resources with optional
project/plan scoping.
Table: ``validation_attachments``
"""
__allow_unmapped__ = True
__tablename__ = "validation_attachments"
# PK: ULID (26-char string)
attachment_id = Column(String(26), primary_key=True)
# FK to tools.name (must be a validation)
validation_name = Column(
String(255),
ForeignKey("tools.name", ondelete="CASCADE"),
nullable=False,
)
# Resource reference
resource_id = Column(String(255), nullable=False)
# Mode: required or informational
mode = Column(String(20), nullable=False)
# Scoping fields
project_name = Column(String(255), nullable=True)
plan_id = Column(String(26), nullable=True)
# Optional arguments
args_json = Column(Text, nullable=True)
# Timestamp
created_at = Column(String(30), nullable=False)
# Relationships
tool = relationship(
"ToolModel",
back_populates="validation_attachments_rel",
)
__table_args__ = (
CheckConstraint(
"mode IN ('required', 'informational')",
name="ck_validation_attachments_mode",
),
Index("ix_validation_attachments_validation_name", "validation_name"),
Index("ix_validation_attachments_resource_id", "resource_id"),
Index("ix_validation_attachments_project_name", "project_name"),
)
# Database initialization functions
def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
"""Initialize the database.
@@ -91,6 +91,9 @@ from cleveragents.infrastructure.database.models import (
ResourceEdgeModel,
ResourceModel,
ResourceTypeModel,
ToolModel,
ToolResourceBindingModel,
ValidationAttachmentModel,
)
@@ -2618,3 +2621,425 @@ class ProjectResourceLinkRepository:
except (OperationalError, SQLAlchemyDatabaseError) as exc:
session.rollback()
raise DatabaseError(f"Failed to remove link '{link_id}': {exc}") from exc
# ---------------------------------------------------------------------------
# Tool Registry Repositories
# ---------------------------------------------------------------------------
class DuplicateToolError(DatabaseError):
"""Raised when creating a tool with a name that already exists."""
def __init__(self, name: str):
super().__init__(f"Tool with name '{name}' already exists")
self.tool_name = name
class ToolInUseError(BusinessRuleViolation):
"""Raised when deleting a tool that still has validation attachments."""
def __init__(self, tool_name: str, attachment_count: int):
super().__init__(
f"Cannot delete tool {tool_name}: "
f"still has {attachment_count} validation attachment(s)"
)
self.tool_name = tool_name
self.attachment_count = attachment_count
class ToolRegistryRepository:
"""Repository for tool and validation persistence.
Uses a session-factory pattern: each public method obtains its own
session from the factory, ensuring proper session lifecycle management.
All mutating methods flush (but do NOT commit); the caller or a
``UnitOfWork`` wrapper is responsible for committing the transaction.
"""
def __init__(self, session_factory: Callable[[], Session]) -> None:
"""Initialise with a callable that returns a new SQLAlchemy Session."""
self._session_factory = session_factory
def _session(self) -> Session:
"""Convenience helper to obtain a session."""
return self._session_factory()
@database_retry
def create(self, tool: Any) -> Any:
"""Persist a new tool domain object.
Args:
tool: A dict-like or domain tool object.
Returns:
The tool after persistence.
Raises:
DuplicateToolError: If a tool with the same name already exists.
DatabaseError: On transient or unexpected DB errors.
"""
session = self._session()
try:
db_model = ToolModel.from_domain(tool)
session.add(db_model)
session.flush()
return tool
except IntegrityError as exc:
session.rollback()
name_str = (
tool.get("name", "")
if isinstance(tool, dict)
else getattr(tool, "name", "")
)
if "UNIQUE" in str(exc).upper() or "unique" in str(exc).lower():
raise DuplicateToolError(name_str) from exc
raise DatabaseError(f"Failed to create tool: {exc}") from exc
except (OperationalError, SQLAlchemyDatabaseError) as exc:
session.rollback()
raise DatabaseError(f"Failed to create tool: {exc}") from exc
@database_retry
def get_by_name(self, name: str) -> Any | None:
"""Retrieve a tool by its namespaced name (PK).
Returns:
The tool as a dict, or ``None`` if not found.
"""
session = self._session()
try:
row = session.query(ToolModel).filter_by(name=name).first()
if row is None:
return None
return row.to_domain()
except (OperationalError, SQLAlchemyDatabaseError) as exc:
raise DatabaseError(f"Failed to get tool {name}: {exc}") from exc
@database_retry
def list_all(
self,
namespace: str | None = None,
tool_type: str | None = None,
source: str | None = None,
) -> list[Any]:
"""List tools with optional filters.
Args:
namespace: Optional namespace filter.
tool_type: Optional tool_type filter (``tool`` or ``validation``).
source: Optional source filter.
Returns:
List of tool dicts.
"""
session = self._session()
try:
query = session.query(ToolModel)
if namespace is not None:
query = query.filter(ToolModel.namespace == namespace)
if tool_type is not None:
query = query.filter(ToolModel.tool_type == tool_type)
if source is not None:
query = query.filter(ToolModel.source == source)
rows = query.order_by(ToolModel.name).all()
return [row.to_domain() for row in rows]
except (OperationalError, SQLAlchemyDatabaseError) as exc:
raise DatabaseError(f"Failed to list tools: {exc}") from exc
@database_retry
def update(self, tool: Any) -> Any:
"""Update all mutable fields of an existing tool.
Replaces child resource bindings entirely.
Args:
tool: The tool dict or object with updated fields.
Returns:
The tool after persistence.
Raises:
DatabaseError: If the tool is not found or a DB error occurs.
"""
session = self._session()
def _get(obj: Any, key: str, default: Any = None) -> Any:
if isinstance(obj, dict):
return obj.get(key, default)
return getattr(obj, key, default)
name_str: str = _get(tool, "name", "")
try:
row = session.query(ToolModel).filter_by(name=name_str).first()
if row is None:
raise DatabaseError(f"Tool {name_str} not found for update")
from datetime import datetime as _dt
row.namespace = name_str.split("/", 1)[0] if "/" in name_str else "" # type: ignore[assignment]
row.short_name = name_str.split("/", 1)[1] if "/" in name_str else name_str # type: ignore[assignment]
row.description = _get(tool, "description", "") # type: ignore[assignment]
row.tool_type = _get(tool, "tool_type", "tool") # type: ignore[assignment]
row.source = _get(tool, "source", "builtin") # type: ignore[assignment]
row.input_schema_json = _get(tool, "input_schema_json") # type: ignore[assignment]
row.output_schema_json = _get(tool, "output_schema_json") # type: ignore[assignment]
row.capability_json = _get(tool, "capability_json") # type: ignore[assignment]
row.lifecycle_json = _get(tool, "lifecycle_json") # type: ignore[assignment]
row.code = _get(tool, "code") # type: ignore[assignment]
row.mcp_server = _get(tool, "mcp_server") # type: ignore[assignment]
row.mcp_tool_name = _get(tool, "mcp_tool_name") # type: ignore[assignment]
row.agent_skill_path = _get(tool, "agent_skill_path") # type: ignore[assignment]
row.timeout = _get(tool, "timeout", 300) # type: ignore[assignment]
row.wraps = _get(tool, "wraps") # type: ignore[assignment]
row.transform = _get(tool, "transform") # type: ignore[assignment]
row.mode = _get(tool, "mode") # type: ignore[assignment]
row.argument_mapping_json = _get(tool, "argument_mapping_json") # type: ignore[assignment]
row.updated_at = _dt.now().isoformat() # type: ignore[assignment]
# Replace child resource bindings
row.resource_bindings_rel.clear() # type: ignore[union-attr]
now_iso = _dt.now().isoformat()
for binding_data in _get(tool, "resource_bindings", []) or []:
row.resource_bindings_rel.append( # type: ignore[union-attr]
ToolResourceBindingModel(
slot_name=_get(binding_data, "slot_name", ""),
resource_type=_get(binding_data, "resource_type", ""),
access_mode=_get(binding_data, "access_mode", "read_only"),
binding_mode=_get(binding_data, "binding_mode", "contextual"),
static_resource=_get(binding_data, "static_resource"),
required=_get(binding_data, "required", True),
description=_get(binding_data, "description"),
created_at=now_iso,
)
)
session.flush()
return tool
except (OperationalError, SQLAlchemyDatabaseError) as exc:
session.rollback()
raise DatabaseError(f"Failed to update tool {name_str}: {exc}") from exc
@database_retry
def delete(self, name: str) -> bool:
"""Delete a tool by namespaced name.
Before deletion, verifies the tool has no validation attachments.
Args:
name: The namespaced name of the tool to delete.
Returns:
``True`` if the tool was deleted.
Raises:
ToolInUseError: If validation attachments still reference this tool.
DatabaseError: On transient or unexpected DB errors.
"""
session = self._session()
try:
# Check for validation attachments
attachment_count: int = (
session.query(ValidationAttachmentModel)
.filter(ValidationAttachmentModel.validation_name == name)
.count()
)
if attachment_count > 0:
raise ToolInUseError(name, attachment_count)
row = session.query(ToolModel).filter_by(name=name).first()
if row is None:
return False
session.delete(row)
session.flush()
return True
except ToolInUseError:
raise
except (OperationalError, SQLAlchemyDatabaseError) as exc:
session.rollback()
raise DatabaseError(f"Failed to delete tool {name}: {exc}") from exc
class ValidationAttachmentRepository:
"""Repository for validation attachment persistence.
Uses a session-factory pattern matching ``ActionRepository``.
"""
def __init__(self, session_factory: Callable[[], Session]) -> None:
"""Initialise with a callable that returns a new SQLAlchemy Session."""
self._session_factory = session_factory
def _session(self) -> Session:
"""Convenience helper to obtain a session."""
return self._session_factory()
@database_retry
def attach(
self,
validation_name: str,
resource_id: str,
mode: str,
project_name: str | None = None,
plan_id: str | None = None,
args: dict[str, Any] | None = None,
) -> Any:
"""Create a validation attachment.
Args:
validation_name: Name of the validation tool.
resource_id: Resource reference string.
mode: ``required`` or ``informational``.
project_name: Optional project scope.
plan_id: Optional plan scope.
args: Optional argument dict (stored as JSON).
Returns:
A dict representing the attachment.
"""
import json as _json
from ulid import ULID as _ULID
session = self._session()
try:
now_iso = datetime.now().isoformat()
attachment_id = str(_ULID())
args_json: str | None = None
if args is not None:
args_json = _json.dumps(args)
model = ValidationAttachmentModel(
attachment_id=attachment_id,
validation_name=validation_name,
resource_id=resource_id,
mode=mode,
project_name=project_name,
plan_id=plan_id,
args_json=args_json,
created_at=now_iso,
)
session.add(model)
session.flush()
return {
"attachment_id": attachment_id,
"validation_name": validation_name,
"resource_id": resource_id,
"mode": mode,
"project_name": project_name,
"plan_id": plan_id,
"args_json": args_json,
"created_at": now_iso,
}
except (OperationalError, SQLAlchemyDatabaseError) as exc:
session.rollback()
raise DatabaseError(f"Failed to attach validation: {exc}") from exc
@database_retry
def detach(self, attachment_id: str) -> bool:
"""Remove a validation attachment by its ULID.
Args:
attachment_id: The ULID of the attachment.
Returns:
``True`` if the attachment was removed.
"""
session = self._session()
try:
row = (
session.query(ValidationAttachmentModel)
.filter_by(attachment_id=attachment_id)
.first()
)
if row is None:
return False
session.delete(row)
session.flush()
return True
except (OperationalError, SQLAlchemyDatabaseError) as exc:
session.rollback()
raise DatabaseError(
f"Failed to detach validation {attachment_id}: {exc}"
) from exc
@database_retry
def list_for_resource(
self,
resource_id: str,
project_name: str | None = None,
plan_id: str | None = None,
) -> list[Any]:
"""List validation attachments for a resource.
Args:
resource_id: The resource reference string.
project_name: Optional project scope filter.
plan_id: Optional plan scope filter.
Returns:
List of attachment dicts.
"""
session = self._session()
try:
query = session.query(ValidationAttachmentModel).filter(
ValidationAttachmentModel.resource_id == resource_id
)
if project_name is not None:
query = query.filter(
ValidationAttachmentModel.project_name == project_name
)
if plan_id is not None:
query = query.filter(ValidationAttachmentModel.plan_id == plan_id)
rows = query.order_by(ValidationAttachmentModel.created_at).all()
results: list[dict[str, Any]] = []
for row in rows:
results.append(
{
"attachment_id": cast(str, row.attachment_id),
"validation_name": cast(str, row.validation_name),
"resource_id": cast(str, row.resource_id),
"mode": cast(str, row.mode),
"project_name": cast("str | None", row.project_name),
"plan_id": cast("str | None", row.plan_id),
"args_json": cast("str | None", row.args_json),
"created_at": cast(str, row.created_at),
}
)
return results
except (OperationalError, SQLAlchemyDatabaseError) as exc:
raise DatabaseError(
f"Failed to list attachments for resource {resource_id}: {exc}"
) from exc
@database_retry
def get_by_id(self, attachment_id: str) -> Any | None:
"""Retrieve an attachment by its ULID.
Returns:
The attachment dict, or ``None`` if not found.
"""
session = self._session()
try:
row = (
session.query(ValidationAttachmentModel)
.filter_by(attachment_id=attachment_id)
.first()
)
if row is None:
return None
return {
"attachment_id": cast(str, row.attachment_id),
"validation_name": cast(str, row.validation_name),
"resource_id": cast(str, row.resource_id),
"mode": cast(str, row.mode),
"project_name": cast("str | None", row.project_name),
"plan_id": cast("str | None", row.plan_id),
"args_json": cast("str | None", row.args_json),
"created_at": cast(str, row.created_at),
}
except (OperationalError, SQLAlchemyDatabaseError) as exc:
raise DatabaseError(
f"Failed to get attachment {attachment_id}: {exc}"
) from exc