test(e2e): add M2 actor + tool source smoke suite
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 16s
CI / lint (pull_request) Successful in 21s
CI / quality (pull_request) Successful in 27s
CI / security (pull_request) Successful in 56s
CI / typecheck (pull_request) Successful in 1m2s
CI / coverage (pull_request) Has been cancelled
CI / unit_tests (pull_request) Has been cancelled
CI / integration_tests (pull_request) Has been cancelled
CI / benchmark-regression (pull_request) Has been cancelled
CI / docker (pull_request) Has been cancelled

Add comprehensive E2E test suite for M2 (Actor Graphs + Tool Sources) epic:

- Behave BDD: 10 scenarios covering actor YAML loading, skill registry,
  tool lifecycle (discover/activate/execute/deactivate), and MCP stub
- Robot Framework: 6 integration tests via CLI helper script
- ASV benchmarks: 12 benchmarks for actor loading, skill registry,
  tool lifecycle, and MCP stub performance baselines
- MCP stub server mock: in-process fake with 3 tools (search/fetch/transform)
- Fixtures: hierarchical graph actor YAML + skill pack with tool refs and
  inline tools
- Docs: updated testing.md with M2 smoke suite section

Closes #169
This commit is contained in:
2026-02-24 21:55:24 +00:00
parent c47e6445d0
commit 093a74953f
9 changed files with 1427 additions and 0 deletions
+203
View File
@@ -0,0 +1,203 @@
"""ASV benchmarks for M2 actor + tool source smoke suite.
Measures baseline runtime for:
- Actor YAML loading from fixture
- Skill registry resolution
- Tool lifecycle overhead (discover/activate/execute/deactivate)
- MCP stub server discovery and invocation
"""
from __future__ import annotations
import importlib
import sys
import tempfile
from pathlib import Path
from typing import Any
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
_MOCKS = str(Path(__file__).resolve().parents[1] / "features" / "mocks")
if _MOCKS not in sys.path:
sys.path.insert(0, _MOCKS)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.actor.loader import ActorLoader # noqa: E402
from cleveragents.actor.schema import ActorConfigSchema # noqa: E402
from cleveragents.domain.models.core.skill import Skill, SkillInlineTool # noqa: E402
from cleveragents.domain.models.core.tool import ( # noqa: E402
Tool,
ToolCapability,
ToolSource,
ToolType,
)
from cleveragents.skills.protocol import SkillDefinition, SkillMetadata # noqa: E402
from cleveragents.skills.registry import SkillRegistry # noqa: E402
from cleveragents.tool.context import ToolExecutionContext # noqa: E402
from cleveragents.tool.lifecycle import ( # noqa: E402
ToolDescriptor,
ToolResult,
ToolRuntime,
)
from mcp_stub_server import McpStubServer # noqa: E402
_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m2"
# ── Actor loading benchmarks ─────────────────────────────────────
class TimeM2ActorLoading:
"""Benchmark actor YAML loading from M2 fixtures."""
def setup(self) -> None:
self.fixture_path = _FIXTURES_DIR / "actors" / "m2_hierarchical_actor.yaml"
def time_load_hierarchical_actor(self) -> None:
ActorConfigSchema.from_yaml_file(self.fixture_path)
def time_discover_from_fixture_dir(self) -> None:
loader = ActorLoader(search_roots=[_FIXTURES_DIR / "actors"])
loader.discover()
def time_actor_loader_cache_hit(self) -> None:
loader = ActorLoader(search_roots=[_FIXTURES_DIR / "actors"])
loader.discover()
loader.discover() # Second call should be a cache hit
# ── Skill registry benchmarks ────────────────────────────────────
class TimeM2SkillRegistry:
"""Benchmark skill registry operations."""
def setup(self) -> None:
self.skill = Skill(
name="m2bench/file-ops",
description="Benchmark skill pack",
tool_refs=["builtin/read_file", "builtin/list_files"],
anonymous_tools=[
SkillInlineTool(
description="Bench echo tool",
source=ToolSource.CUSTOM,
code="def echo(t: str) -> str: return t",
timeout=300,
),
],
)
self.metadata = SkillMetadata.from_skill(self.skill)
self.defn = SkillDefinition(skill=self.skill, metadata=self.metadata)
def time_register_skill(self) -> None:
registry = SkillRegistry()
registry.register(self.defn)
def time_resolve_tools(self) -> None:
registry = SkillRegistry()
registry.register(self.defn)
registry.resolve_tools(self.skill.name)
def time_list_all_skills(self) -> None:
registry = SkillRegistry()
registry.register(self.defn)
registry.list_all()
# ── Tool lifecycle benchmarks ────────────────────────────────────
class _BenchToolInstance:
"""Minimal tool instance for benchmark."""
def __init__(self, name: str) -> None:
self._name = name
def discover(self) -> ToolDescriptor:
return ToolDescriptor(
name=self._name, description="bench tool", source="custom"
)
def activate(self, ctx: ToolExecutionContext) -> None:
pass
def execute(self, params: dict[str, Any], ctx: ToolExecutionContext) -> ToolResult:
return ToolResult(success=True, data={"echo": params.get("text", "")})
def deactivate(self, ctx: ToolExecutionContext) -> None:
pass
class TimeM2ToolLifecycle:
"""Benchmark tool lifecycle overhead."""
def setup(self) -> None:
self.name = "m2bench/echo"
self.tool = Tool(
name=self.name,
description="Benchmark tool",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
capability=ToolCapability(
writes=False, read_only=True, checkpointable=False
),
timeout=300,
)
self.instance = _BenchToolInstance(self.name)
def time_register_and_discover(self) -> None:
runtime = ToolRuntime()
runtime.register_tool(self.tool, self.instance)
runtime.discover(self.name)
def time_full_lifecycle(self) -> None:
runtime = ToolRuntime()
runtime.register_tool(self.tool, self.instance)
ctx = ToolExecutionContext(plan_id="bench-plan")
runtime.activate(self.name, ctx)
runtime.execute(self.name, {"text": "bench"}, ctx)
runtime.deactivate(self.name, ctx)
def time_execute_only(self) -> None:
runtime = ToolRuntime()
runtime.register_tool(self.tool, self.instance)
ctx = ToolExecutionContext(plan_id="bench-plan")
runtime.execute(self.name, {"text": "bench"}, ctx)
# ── MCP stub benchmarks ─────────────────────────────────────────
class TimeM2McpStub:
"""Benchmark MCP stub server discovery and invocation."""
def setup(self) -> None:
self.server = McpStubServer()
self.server.start()
def teardown(self) -> None:
self.server.stop()
def time_discover_tools(self) -> None:
self.server.discover()
def time_invoke_search(self) -> None:
self.server.invoke("mcp/search", {"query": "benchmark"})
def time_invoke_fetch(self) -> None:
self.server.invoke("mcp/fetch", {"url": "http://example.com"})
def time_invoke_transform(self) -> None:
self.server.invoke("mcp/transform", {"data": "hello", "format": "upper"})
# Module-level instances for ASV discovery
time_actor = TimeM2ActorLoading()
time_skill = TimeM2SkillRegistry()
time_tool = TimeM2ToolLifecycle()
time_mcp = TimeM2McpStub()
+58
View File
@@ -714,3 +714,61 @@ nox -s benchmark
| `benchmarks/scale_fixture_bench.py` | ASV | 6 | Performance benchmarks for fixture processing |
For full scale testing documentation, see `docs/development/scale_testing.md`.
## M2 Actor + Tool Source Smoke Suite
The M2 smoke suite validates the foundation for the Actor Graphs + Tool Sources epic (#356). It covers actor YAML loading, skill registry, tool lifecycle, and MCP stub tool discovery.
### Fixtures
M2 fixtures live in `features/fixtures/m2/`:
| Fixture File | Description |
|---|---|
| `m2_hierarchical_actor.yaml` | Graph-type actor with planner → executor → reviewer nodes |
| `m2_skill_pack.yaml` | Skill pack with tool references and an inline tool |
### MCP Stub Server
The MCP stub server (`features/mocks/mcp_stub_server.py`) simulates MCP tool discovery and invocation in-process without requiring the real MCP adapter (#159). It exposes three deterministic tools: `mcp/search`, `mcp/fetch`, and `mcp/transform`.
Usage in tests:
```python
from features.mocks.mcp_stub_server import McpStubServer
server = McpStubServer()
server.start()
tools = server.discover() # Returns 3 stub tools
result = server.invoke("mcp/search", {"query": "hello"})
server.stop()
```
### Test Suites
| Suite | Framework | Scenarios | Description |
|-------|-----------|-----------|-------------|
| `features/m2_actor_tool_smoke.feature` | Behave | 11 | Actor loading, skill registry, tool lifecycle, MCP stub |
| `robot/m2_actor_tool_smoke.robot` | Robot | 6 | CLI-level smoke for actor/skill/tool/MCP operations |
| `benchmarks/m2_actor_tool_smoke_bench.py` | ASV | 12 | Baseline runtime for actor, skill, tool, and MCP operations |
### Running the M2 Smoke Suite
```bash
# Behave only (M2 feature)
nox -s unit_tests -- features/m2_actor_tool_smoke.feature
# Robot only (M2 suite)
nox -s integration_tests -- --suite robot/m2_actor_tool_smoke.robot
# Benchmarks
nox -s benchmark
```
### What Is NOT Tested (deferred to dependent issues)
- Real MCP adapter integration (#159, assigned to Aditya)
- Agent skills loader (#160, assigned to Aditya)
- Hierarchical actor compiler (#158, assigned to Jeff)
The MCP stub server provides a test seam that can be replaced with the real adapter once #159 lands.
@@ -0,0 +1,44 @@
name: m2test/hierarchical-workflow
type: graph
description: M2 smoke test hierarchical actor with planner and executor nodes
version: "1.0"
model: gpt-4
context_view: full
memory:
enabled: true
max_messages: 100
context:
include_files:
- README.md
exclude_patterns:
- "**/__pycache__/**"
route:
nodes:
- id: planner
type: agent
name: Planner
description: Plans the next steps
config:
model: gpt-4
prompt: You are a task planner
- id: executor
type: tool
name: Executor
description: Executes file operations
config:
tool_name: builtin/read_file
- id: reviewer
type: agent
name: Reviewer
description: Reviews output quality
config:
model: gpt-4
prompt: You are a code reviewer
edges:
- from_node: planner
to_node: executor
- from_node: executor
to_node: reviewer
entry_node: planner
exit_nodes:
- reviewer
+23
View File
@@ -0,0 +1,23 @@
name: m2test/file-ops-pack
description: M2 smoke test skill pack with file operation tools
tools:
- name: builtin/read_file
description: Read contents of a file
- name: builtin/list_files
description: List files in a directory
inline_tools:
- name: m2_echo
description: Echo input back for testing
source: custom
code: |
def m2_echo(text: str) -> str:
return f"echo: {text}"
input_schema:
type: object
properties:
text:
type: string
required:
- text
writes: false
checkpointable: false
+86
View File
@@ -0,0 +1,86 @@
Feature: M2 actor + tool source smoke suite
As a CleverAgents developer
I want to verify that actor loading, skill registry, tool lifecycle,
and MCP stub tool discovery work end-to-end
So that the M2 (Actor Graphs + Tool Sources) foundation is solid
# Actor YAML loading
Scenario: Load hierarchical actor from YAML fixture
Given I load the M2 hierarchical actor fixture
Then the M2 actor name should be "m2test/hierarchical-workflow"
And the M2 actor type should be "graph"
And the M2 actor should have 3 graph nodes
And the M2 actor entry node should be "planner"
And the M2 actor exit nodes should include "reviewer"
Scenario: Load hierarchical actor via ActorLoader from fixture directory
Given I create an M2 actor loader from the fixture directory
When I run M2 actor discovery
Then the M2 loader should find 1 actors
And the M2 loader should contain actor "m2test/hierarchical-workflow"
# ── Skill pack loading and registry ──────────────────────────────
Scenario: Load skill pack from YAML fixture
Given I load the M2 skill pack fixture
Then the M2 skill name should be "m2test/file-ops-pack"
And the M2 skill should have 2 tool references
And the M2 skill should have 1 inline tools
Scenario: Register skill pack and verify tool resolution
Given I register the M2 skill pack in a skill registry
When I resolve tools for M2 skill "m2test/file-ops-pack"
Then the M2 resolved tools should include a tool_ref entry
And the M2 resolved tools should include an inline entry
Scenario: Skill registry lists registered skills correctly
Given I register the M2 skill pack in a skill registry
When I list all M2 skills in the registry
Then the M2 skill list should contain "m2test/file-ops-pack"
And the M2 skill list should have 1 entries
Scenario: Invoke built-in tool through skill context
Given I create an M2 skill context with sandbox
When I register a tool invocation for "builtin/read_file" in the M2 context
Then the M2 skill context should have 1 invocation records
# ── Tool lifecycle ───────────────────────────────────────────────
Scenario: Tool lifecycle discover activate execute deactivate
Given I create an M2 tool runtime with a mock tool "m2test/echo"
And I create an M2 execution context with plan "m2-plan-001"
When I discover M2 tool "m2test/echo"
Then the M2 discovered descriptor name should be "m2test/echo"
When I activate M2 tool "m2test/echo"
And I execute M2 tool "m2test/echo" with params text "hello"
Then the M2 execution result should be successful
When I deactivate M2 tool "m2test/echo"
Then the M2 tool should be deactivated
Scenario: Tool registry registers and retrieves tools
Given I create an M2 tool registry
When I register an M2 tool spec "m2test/file-reader"
Then getting M2 tool "m2test/file-reader" should return a spec
And listing M2 tools should include "m2test/file-reader"
# ── MCP stub tool discovery and invocation ───────────────────────
Scenario: MCP stub tool discovery and invocation
Given I start the M2 MCP stub server
When I discover tools from the M2 MCP stub
Then the M2 stub should expose 3 tools
And the M2 stub tools should include "mcp/search"
And the M2 stub tools should include "mcp/fetch"
And the M2 stub tools should include "mcp/transform"
When I invoke M2 MCP stub tool "mcp/search" with query "test"
Then the M2 MCP stub result should contain results
And I stop the M2 MCP stub server
Scenario: MCP stub server lifecycle guards
Given I have an M2 MCP stub server that is stopped
When I try to discover tools from the stopped M2 stub
Then an M2 RuntimeError should be raised
When I start and invoke M2 MCP stub tool "mcp/fetch" with url "http://example.com"
Then the M2 MCP stub fetch result should have status 200
And the M2 MCP stub invocation log should have 1 entries
+194
View File
@@ -0,0 +1,194 @@
"""MCP stub server mock for M2 actor + tool source smoke tests.
Simulates MCP tool discovery and invocation in-process without requiring
a real MCP adapter. Returns deterministic tool descriptors and execution
results for Behave-level testing.
This mock lives in ``features/mocks/`` per ADR-022 (no mocks in src/).
"""
from __future__ import annotations
from typing import Any, ClassVar
class McpStubTool:
"""Descriptor for a single tool exposed by the MCP stub server."""
def __init__(
self,
name: str,
description: str,
input_schema: dict[str, Any] | None = None,
output_schema: dict[str, Any] | None = None,
) -> None:
self.name = name
self.description = description
self.input_schema = input_schema or {
"type": "object",
"properties": {"query": {"type": "string"}},
}
self.output_schema = output_schema or {
"type": "object",
"properties": {"result": {"type": "string"}},
}
class McpStubServer:
"""In-process MCP stub server for deterministic testing.
Provides ``discover()`` and ``invoke()`` methods that mirror the
future MCP adapter interface (#159) without requiring network I/O.
Usage::
server = McpStubServer()
server.start()
tools = server.discover()
result = server.invoke("mcp/search", {"query": "hello"})
server.stop()
"""
#: Default tools returned by ``discover()``.
DEFAULT_TOOLS: ClassVar[list[McpStubTool]] = [
McpStubTool(
name="mcp/search",
description="Stub MCP search tool",
input_schema={
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
output_schema={
"type": "object",
"properties": {
"results": {"type": "array", "items": {"type": "string"}}
},
},
),
McpStubTool(
name="mcp/fetch",
description="Stub MCP fetch tool",
input_schema={
"type": "object",
"properties": {"url": {"type": "string"}},
"required": ["url"],
},
output_schema={
"type": "object",
"properties": {
"content": {"type": "string"},
"status": {"type": "integer"},
},
},
),
McpStubTool(
name="mcp/transform",
description="Stub MCP transform tool",
input_schema={
"type": "object",
"properties": {
"data": {"type": "string"},
"format": {"type": "string"},
},
"required": ["data"],
},
output_schema={
"type": "object",
"properties": {"transformed": {"type": "string"}},
},
),
]
def __init__(self, tools: list[McpStubTool] | None = None) -> None:
self._tools = tools or list(self.DEFAULT_TOOLS)
self._running = False
self._invocation_log: list[dict[str, Any]] = []
@property
def is_running(self) -> bool:
"""Whether the stub server is currently started."""
return self._running
@property
def invocation_log(self) -> list[dict[str, Any]]:
"""Log of all invocations since the last ``start()``."""
return list(self._invocation_log)
def start(self) -> None:
"""Start the stub server (in-process, no network)."""
self._running = True
self._invocation_log = []
def stop(self) -> None:
"""Stop the stub server."""
self._running = False
def discover(self) -> list[McpStubTool]:
"""Return the list of available tools.
Raises:
RuntimeError: If the server is not running.
"""
if not self._running:
msg = "McpStubServer is not running. Call start() first."
raise RuntimeError(msg)
return list(self._tools)
def invoke(self, tool_name: str, params: dict[str, Any]) -> dict[str, Any]:
"""Invoke a stub tool by name with deterministic results.
Raises:
RuntimeError: If the server is not running.
KeyError: If the tool name is not found.
"""
if not self._running:
msg = "McpStubServer is not running. Call start() first."
raise RuntimeError(msg)
tool = self._find_tool(tool_name)
if tool is None:
msg = f"Tool '{tool_name}' not found in MCP stub server."
raise KeyError(msg)
# Generate deterministic result based on tool name
result = self._generate_result(tool, params)
self._invocation_log.append(
{
"tool_name": tool_name,
"params": params,
"result": result,
}
)
return result
def _find_tool(self, name: str) -> McpStubTool | None:
"""Find a tool by name."""
for tool in self._tools:
if tool.name == name:
return tool
return None
def _generate_result(
self, tool: McpStubTool, params: dict[str, Any]
) -> dict[str, Any]:
"""Generate a deterministic result for a tool invocation."""
if tool.name == "mcp/search":
query = params.get("query", "")
return {
"results": [f"result_1_for_{query}", f"result_2_for_{query}"],
}
if tool.name == "mcp/fetch":
url = params.get("url", "")
return {
"content": f"<html>stub content for {url}</html>",
"status": 200,
}
if tool.name == "mcp/transform":
data = params.get("data", "")
fmt = params.get("format", "upper")
transformed = data.upper() if fmt == "upper" else data.lower()
return {"transformed": transformed}
# Generic fallback
return {"result": f"stub result from {tool.name}"}
+515
View File
@@ -0,0 +1,515 @@
"""Step definitions for m2_actor_tool_smoke.feature.
All step names are prefixed with 'M2' to avoid AmbiguousStep conflicts
with existing actor/skill/tool step files.
"""
from __future__ import annotations
import tempfile
from pathlib import Path
from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.actor.loader import ActorLoader
from cleveragents.actor.schema import ActorConfigSchema
from cleveragents.domain.models.core.skill import (
Skill,
SkillInlineTool,
)
from cleveragents.domain.models.core.tool import (
Tool,
ToolCapability,
ToolSource,
ToolType,
)
from cleveragents.skills.context import SkillContext
from cleveragents.skills.protocol import SkillDefinition, SkillMetadata
from cleveragents.skills.registry import SkillRegistry
from cleveragents.skills.schema import SkillConfigSchema
from cleveragents.tool.context import ToolExecutionContext
from cleveragents.tool.lifecycle import (
ToolDescriptor,
ToolResult,
ToolRuntime,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runtime import ToolSpec
# ── Fixture paths ──────────────────────────────────────────────────
_FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures" / "m2"
_ACTORS_DIR = _FIXTURES_DIR / "actors"
_HIERARCHICAL_ACTOR = _ACTORS_DIR / "m2_hierarchical_actor.yaml"
_SKILL_PACK = _FIXTURES_DIR / "m2_skill_pack.yaml"
# ── Mock ToolInstance for lifecycle tests ──────────────────────────
class _M2MockToolInstance:
"""Minimal ToolInstance for M2 lifecycle testing."""
def __init__(self, name: str) -> None:
self._name = name
self._activated = False
self._deactivated = False
self._execute_count = 0
def discover(self) -> ToolDescriptor:
return ToolDescriptor(
name=self._name,
description=f"M2 mock tool {self._name}",
source="custom",
)
def activate(self, ctx: ToolExecutionContext) -> None:
self._activated = True
def execute(self, params: dict[str, Any], ctx: ToolExecutionContext) -> ToolResult:
self._execute_count += 1
return ToolResult(success=True, data={"echo": params.get("text", "")})
def deactivate(self, ctx: ToolExecutionContext) -> None:
self._deactivated = True
@property
def is_activated(self) -> bool:
return self._activated
@property
def is_deactivated(self) -> bool:
return self._deactivated
# ════════════════════════════════════════════════════════════════════
# Actor YAML loading
# ════════════════════════════════════════════════════════════════════
@given("I load the M2 hierarchical actor fixture")
def step_m2_load_hierarchical_actor(context: Context) -> None:
context._m2_actor = ActorConfigSchema.from_yaml_file(_HIERARCHICAL_ACTOR)
@then('the M2 actor name should be "{name}"')
def step_m2_actor_name(context: Context, name: str) -> None:
actor: ActorConfigSchema = context._m2_actor
assert actor.name == name, f"Expected '{name}', got '{actor.name}'"
@then('the M2 actor type should be "{actor_type}"')
def step_m2_actor_type(context: Context, actor_type: str) -> None:
actor: ActorConfigSchema = context._m2_actor
assert actor.type.value == actor_type, (
f"Expected type '{actor_type}', got '{actor.type.value}'"
)
@then("the M2 actor should have {count:d} graph nodes")
def step_m2_actor_node_count(context: Context, count: int) -> None:
actor: ActorConfigSchema = context._m2_actor
assert actor.route is not None, "Expected a route definition"
assert len(actor.route.nodes) == count, (
f"Expected {count} nodes, got {len(actor.route.nodes)}"
)
@then('the M2 actor entry node should be "{entry}"')
def step_m2_actor_entry_node(context: Context, entry: str) -> None:
actor: ActorConfigSchema = context._m2_actor
assert actor.route is not None
assert actor.route.entry_node == entry, (
f"Expected entry '{entry}', got '{actor.route.entry_node}'"
)
@then('the M2 actor exit nodes should include "{exit_node}"')
def step_m2_actor_exit_nodes(context: Context, exit_node: str) -> None:
actor: ActorConfigSchema = context._m2_actor
assert actor.route is not None
assert exit_node in actor.route.exit_nodes, (
f"Expected '{exit_node}' in exit nodes: {actor.route.exit_nodes}"
)
# ── Actor loader discovery from fixture dir ───────────────────────
@given("I create an M2 actor loader from the fixture directory")
def step_m2_create_loader(context: Context) -> None:
context._m2_loader = ActorLoader(search_roots=[_ACTORS_DIR])
@when("I run M2 actor discovery")
def step_m2_run_discovery(context: Context) -> None:
loader: ActorLoader = context._m2_loader
context._m2_discovered = loader.discover()
@then("the M2 loader should find {count:d} actors")
def step_m2_loader_count(context: Context, count: int) -> None:
loader: ActorLoader = context._m2_loader
actors = loader.list_actors()
assert len(actors) == count, (
f"Expected {count} actors, got {len(actors)}: {[a.name for a in actors]}"
)
@then('the M2 loader should contain actor "{name}"')
def step_m2_loader_contains(context: Context, name: str) -> None:
loader: ActorLoader = context._m2_loader
config = loader.get(name)
assert config is not None, f"Actor '{name}' not found in M2 loader"
# ════════════════════════════════════════════════════════════════════
# Skill pack loading and registry
# ════════════════════════════════════════════════════════════════════
@given("I load the M2 skill pack fixture")
def step_m2_load_skill_pack(context: Context) -> None:
context._m2_skill_config = SkillConfigSchema.from_yaml_file(_SKILL_PACK)
@then('the M2 skill name should be "{name}"')
def step_m2_skill_name(context: Context, name: str) -> None:
cfg: SkillConfigSchema = context._m2_skill_config
assert cfg.name == name, f"Expected '{name}', got '{cfg.name}'"
@then("the M2 skill should have {count:d} tool references")
def step_m2_skill_tool_refs(context: Context, count: int) -> None:
cfg: SkillConfigSchema = context._m2_skill_config
assert len(cfg.tools) == count, f"Expected {count} tool refs, got {len(cfg.tools)}"
@then("the M2 skill should have {count:d} inline tools")
def step_m2_skill_inline_tools(context: Context, count: int) -> None:
cfg: SkillConfigSchema = context._m2_skill_config
assert len(cfg.inline_tools) == count, (
f"Expected {count} inline tools, got {len(cfg.inline_tools)}"
)
def _build_m2_skill_definition() -> SkillDefinition:
"""Build a SkillDefinition from the M2 fixture for registry tests."""
skill = Skill(
name="m2test/file-ops-pack",
description="M2 smoke test skill pack with file operation tools",
tool_refs=["builtin/read_file", "builtin/list_files"],
anonymous_tools=[
SkillInlineTool(
description="Echo input back for testing",
source=ToolSource.CUSTOM,
code="def m2_echo(text: str) -> str:\n return f'echo: {text}'",
timeout=300,
),
],
)
metadata = SkillMetadata.from_skill(skill)
return SkillDefinition(
skill=skill,
metadata=metadata,
)
@given("I register the M2 skill pack in a skill registry")
def step_m2_register_skill_pack(context: Context) -> None:
registry = SkillRegistry()
defn = _build_m2_skill_definition()
registry.register(defn)
context._m2_skill_registry = registry
@when('I resolve tools for M2 skill "{name}"')
def step_m2_resolve_tools(context: Context, name: str) -> None:
registry: SkillRegistry = context._m2_skill_registry
context._m2_resolved_tools = registry.resolve_tools(name)
@then("the M2 resolved tools should include a tool_ref entry")
def step_m2_resolved_has_tool_ref(context: Context) -> None:
entries = context._m2_resolved_tools
has_ref = any(not e.is_inline for e in entries)
assert has_ref, "Expected at least one tool_ref entry in resolved tools"
@then("the M2 resolved tools should include an inline entry")
def step_m2_resolved_has_inline(context: Context) -> None:
entries = context._m2_resolved_tools
has_inline = any(e.is_inline for e in entries)
assert has_inline, "Expected at least one inline entry in resolved tools"
@when("I list all M2 skills in the registry")
def step_m2_list_skills(context: Context) -> None:
registry: SkillRegistry = context._m2_skill_registry
context._m2_skill_list = registry.list_all()
@then('the M2 skill list should contain "{name}"')
def step_m2_skill_list_contains(context: Context, name: str) -> None:
skill_list = context._m2_skill_list
names = [m.name for m in skill_list]
assert name in names, f"Expected '{name}' in {names}"
@then("the M2 skill list should have {count:d} entries")
def step_m2_skill_list_count(context: Context, count: int) -> None:
skill_list = context._m2_skill_list
assert len(skill_list) == count, f"Expected {count} entries, got {len(skill_list)}"
# ── Skill context invocation ──────────────────────────────────────
@given("I create an M2 skill context with sandbox")
def step_m2_create_skill_context(context: Context) -> None:
tmp = Path(tempfile.mkdtemp(prefix="m2_skill_ctx_"))
context._m2_skill_ctx = SkillContext(
plan_id="m2-plan-001",
project_id="m2-project-001",
sandbox_path=tmp,
)
@when('I register a tool invocation for "{tool_name}" in the M2 context')
def step_m2_register_invocation(context: Context, tool_name: str) -> None:
ctx: SkillContext = context._m2_skill_ctx
ctx.register_tool_invocation(
tool_name=tool_name,
input_data={"path": "test.txt"},
output_data={"content": "hello"},
duration_ms=1.5,
)
@then("the M2 skill context should have {count:d} invocation records")
def step_m2_skill_ctx_invocations(context: Context, count: int) -> None:
ctx: SkillContext = context._m2_skill_ctx
assert len(ctx.change_tracker) == count, (
f"Expected {count} records, got {len(ctx.change_tracker)}"
)
# ════════════════════════════════════════════════════════════════════
# Tool lifecycle
# ════════════════════════════════════════════════════════════════════
@given('I create an M2 tool runtime with a mock tool "{name}"')
def step_m2_create_tool_runtime(context: Context, name: str) -> None:
tool = Tool(
name=name,
description=f"M2 mock tool {name}",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
capability=ToolCapability(writes=False, read_only=True, checkpointable=False),
timeout=300,
)
instance = _M2MockToolInstance(name)
runtime = ToolRuntime()
runtime.register_tool(tool, instance)
context._m2_runtime = runtime
context._m2_mock_instance = instance
@given('I create an M2 execution context with plan "{plan_id}"')
def step_m2_create_exec_ctx(context: Context, plan_id: str) -> None:
context._m2_exec_ctx = ToolExecutionContext(plan_id=plan_id)
@when('I discover M2 tool "{name}"')
def step_m2_discover_tool(context: Context, name: str) -> None:
runtime: ToolRuntime = context._m2_runtime
context._m2_descriptor = runtime.discover(name)
@then('the M2 discovered descriptor name should be "{name}"')
def step_m2_descriptor_name(context: Context, name: str) -> None:
desc: ToolDescriptor = context._m2_descriptor
assert desc.name == name, f"Expected '{name}', got '{desc.name}'"
@when('I activate M2 tool "{name}"')
def step_m2_activate_tool(context: Context, name: str) -> None:
runtime: ToolRuntime = context._m2_runtime
ctx: ToolExecutionContext = context._m2_exec_ctx
runtime.activate(name, ctx)
@when('I execute M2 tool "{name}" with params text "{text}"')
def step_m2_execute_tool(context: Context, name: str, text: str) -> None:
runtime: ToolRuntime = context._m2_runtime
ctx: ToolExecutionContext = context._m2_exec_ctx
context._m2_exec_result = runtime.execute(name, {"text": text}, ctx)
@then("the M2 execution result should be successful")
def step_m2_exec_result_success(context: Context) -> None:
result: ToolResult = context._m2_exec_result
assert result.success, f"Expected success, got error: {result.error}"
@when('I deactivate M2 tool "{name}"')
def step_m2_deactivate_tool(context: Context, name: str) -> None:
runtime: ToolRuntime = context._m2_runtime
ctx: ToolExecutionContext = context._m2_exec_ctx
runtime.deactivate(name, ctx)
@then("the M2 tool should be deactivated")
def step_m2_tool_deactivated(context: Context) -> None:
instance: _M2MockToolInstance = context._m2_mock_instance
assert instance.is_deactivated, "Expected mock tool to be deactivated"
# ── Tool registry ─────────────────────────────────────────────────
@given("I create an M2 tool registry")
def step_m2_create_tool_registry(context: Context) -> None:
context._m2_tool_registry = ToolRegistry()
@when('I register an M2 tool spec "{name}"')
def step_m2_register_tool_spec(context: Context, name: str) -> None:
registry: ToolRegistry = context._m2_tool_registry
spec = ToolSpec(
name=name,
description=f"M2 test tool {name}",
handler=lambda **kwargs: {"ok": True},
)
registry.register(spec)
@then('getting M2 tool "{name}" should return a spec')
def step_m2_get_tool_spec(context: Context, name: str) -> None:
registry: ToolRegistry = context._m2_tool_registry
spec = registry.get(name)
assert spec is not None, f"Expected spec for '{name}', got None"
@then('listing M2 tools should include "{name}"')
def step_m2_list_tools_include(context: Context, name: str) -> None:
registry: ToolRegistry = context._m2_tool_registry
specs = registry.list_tools()
names = [s.name for s in specs]
assert name in names, f"Expected '{name}' in {names}"
# ════════════════════════════════════════════════════════════════════
# MCP stub tool discovery and invocation
# ════════════════════════════════════════════════════════════════════
def _get_mcp_stub_class() -> type:
"""Lazily import McpStubServer to avoid top-level sys.path mutation."""
import sys as _sys
_mocks_dir = str(Path(__file__).resolve().parent.parent / "mocks")
if _mocks_dir not in _sys.path:
_sys.path.insert(0, _mocks_dir)
from mcp_stub_server import McpStubServer
return McpStubServer
@given("I start the M2 MCP stub server")
def step_m2_start_mcp_stub(context: Context) -> None:
cls = _get_mcp_stub_class()
server = cls()
server.start()
context._m2_mcp_stub = server
@when("I discover tools from the M2 MCP stub")
def step_m2_discover_mcp_tools(context: Context) -> None:
server = context._m2_mcp_stub
context._m2_mcp_tools = server.discover()
@then("the M2 stub should expose {count:d} tools")
def step_m2_stub_tool_count(context: Context, count: int) -> None:
tools = context._m2_mcp_tools
assert len(tools) == count, f"Expected {count} tools, got {len(tools)}"
@then('the M2 stub tools should include "{name}"')
def step_m2_stub_tools_include(context: Context, name: str) -> None:
tools = context._m2_mcp_tools
names = [t.name for t in tools]
assert name in names, f"Expected '{name}' in {names}"
@when('I invoke M2 MCP stub tool "{name}" with query "{query}"')
def step_m2_invoke_mcp_search(context: Context, name: str, query: str) -> None:
server = context._m2_mcp_stub
context._m2_mcp_result = server.invoke(name, {"query": query})
@then("the M2 MCP stub result should contain results")
def step_m2_mcp_result_has_results(context: Context) -> None:
result = context._m2_mcp_result
assert "results" in result, f"Expected 'results' key in {result}"
assert len(result["results"]) > 0, "Expected non-empty results"
@then("I stop the M2 MCP stub server")
def step_m2_stop_mcp_stub(context: Context) -> None:
server = context._m2_mcp_stub
server.stop()
assert not server.is_running, "Expected MCP stub to be stopped"
# ── MCP stub lifecycle guards ─────────────────────────────────────
@given("I have an M2 MCP stub server that is stopped")
def step_m2_mcp_stub_stopped(context: Context) -> None:
cls = _get_mcp_stub_class()
context._m2_mcp_stub = cls()
@when("I try to discover tools from the stopped M2 stub")
def step_m2_discover_stopped(context: Context) -> None:
server = context._m2_mcp_stub
try:
server.discover()
context._m2_mcp_error = None
except RuntimeError as exc:
context._m2_mcp_error = exc
@then("an M2 RuntimeError should be raised")
def step_m2_runtime_error_raised(context: Context) -> None:
err = context._m2_mcp_error
assert err is not None, "Expected RuntimeError but none was raised"
assert isinstance(err, RuntimeError)
@when('I start and invoke M2 MCP stub tool "{name}" with url "{url}"')
def step_m2_start_and_invoke_fetch(context: Context, name: str, url: str) -> None:
server = context._m2_mcp_stub
server.start()
context._m2_mcp_result = server.invoke(name, {"url": url})
@then("the M2 MCP stub fetch result should have status {status:d}")
def step_m2_mcp_fetch_status(context: Context, status: int) -> None:
result = context._m2_mcp_result
assert result.get("status") == status, (
f"Expected status {status}, got {result.get('status')}"
)
@then("the M2 MCP stub invocation log should have {count:d} entries")
def step_m2_mcp_invocation_log(context: Context, count: int) -> None:
server = context._m2_mcp_stub
log = server.invocation_log
assert len(log) == count, f"Expected {count} entries, got {len(log)}"
+236
View File
@@ -0,0 +1,236 @@
"""Robot Framework helper for M2 actor + tool source smoke tests.
Provides a CLI-style interface for Robot to invoke actor loading, skill
registry, tool lifecycle, and MCP stub operations.
Usage:
python robot/helper_m2_actor_tool_smoke.py actor-load-fixture
python robot/helper_m2_actor_tool_smoke.py actor-discover
python robot/helper_m2_actor_tool_smoke.py skill-load-fixture
python robot/helper_m2_actor_tool_smoke.py skill-registry
python robot/helper_m2_actor_tool_smoke.py tool-lifecycle
python robot/helper_m2_actor_tool_smoke.py mcp-stub
"""
from __future__ import annotations
import sys
from pathlib import Path
from typing import Any
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
_MOCKS = str(Path(__file__).resolve().parents[1] / "features" / "mocks")
if _MOCKS not in sys.path:
sys.path.insert(0, _MOCKS)
_FIXTURES_DIR = Path(__file__).resolve().parents[1] / "features" / "fixtures" / "m2"
def main() -> int:
if len(sys.argv) < 2:
print("Usage: helper_m2_actor_tool_smoke.py <command>")
return 1
command = sys.argv[1]
dispatch = {
"actor-load-fixture": _actor_load_fixture,
"actor-discover": _actor_discover,
"skill-load-fixture": _skill_load_fixture,
"skill-registry": _skill_registry,
"tool-lifecycle": _tool_lifecycle,
"mcp-stub": _mcp_stub,
}
handler = dispatch.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
return handler()
def _actor_load_fixture() -> int:
from cleveragents.actor.schema import ActorConfigSchema
path = _FIXTURES_DIR / "actors" / "m2_hierarchical_actor.yaml"
actor = ActorConfigSchema.from_yaml_file(path)
print(f"actor-name: {actor.name}")
print(f"actor-type: {actor.type.value}")
if actor.route:
print(f"node-count: {len(actor.route.nodes)}")
print(f"entry-node: {actor.route.entry_node}")
return 0
def _actor_discover() -> int:
from cleveragents.actor.loader import ActorLoader
loader = ActorLoader(search_roots=[_FIXTURES_DIR / "actors"])
actors = loader.discover()
for actor in actors:
print(f"actor-loaded: {actor.name}")
print(f"actor-count: {len(actors)}")
return 0
def _skill_load_fixture() -> int:
from cleveragents.skills.schema import SkillConfigSchema
path = _FIXTURES_DIR / "m2_skill_pack.yaml"
cfg = SkillConfigSchema.from_yaml_file(path)
print(f"skill-name: {cfg.name}")
print(f"tool-ref-count: {len(cfg.tools)}")
print(f"inline-tool-count: {len(cfg.inline_tools)}")
return 0
def _skill_registry() -> int:
from cleveragents.domain.models.core.skill import Skill, SkillInlineTool
from cleveragents.domain.models.core.tool import ToolSource
from cleveragents.skills.protocol import SkillDefinition, SkillMetadata
from cleveragents.skills.registry import SkillRegistry
skill = Skill(
name="m2test/file-ops-pack",
description="M2 smoke test skill pack with file operation tools",
tool_refs=["builtin/read_file", "builtin/list_files"],
anonymous_tools=[
SkillInlineTool(
description="Echo input back for testing",
source=ToolSource.CUSTOM,
code="def m2_echo(text: str) -> str:\n return f'echo: {text}'",
timeout=300,
),
],
)
metadata = SkillMetadata.from_skill(skill)
defn = SkillDefinition(skill=skill, metadata=metadata)
registry = SkillRegistry()
registry.register(defn)
print(f"skill-registered: {skill.name}")
resolved = registry.resolve_tools(skill.name)
has_ref = any(not e.is_inline for e in resolved)
has_inline = any(e.is_inline for e in resolved)
if has_ref and has_inline:
print("resolved-tools-ok")
else:
print(f"resolved-tools-fail: ref={has_ref} inline={has_inline}")
return 1
return 0
def _tool_lifecycle() -> int:
from cleveragents.domain.models.core.tool import (
Tool,
ToolCapability,
ToolSource,
ToolType,
)
from cleveragents.tool.context import ToolExecutionContext
from cleveragents.tool.lifecycle import ToolDescriptor, ToolResult, ToolRuntime
class MockInstance:
def __init__(self, name: str) -> None:
self._name = name
self.activated = False
self.deactivated = False
def discover(self) -> ToolDescriptor:
return ToolDescriptor(
name=self._name, description=f"mock {self._name}", source="custom"
)
def activate(self, ctx: ToolExecutionContext) -> None:
self.activated = True
def execute(
self, params: dict[str, Any], ctx: ToolExecutionContext
) -> ToolResult:
return ToolResult(success=True, data={"echo": params.get("text", "")})
def deactivate(self, ctx: ToolExecutionContext) -> None:
self.deactivated = True
name = "m2test/echo"
tool = Tool(
name=name,
description="M2 mock tool",
source=ToolSource.BUILTIN,
tool_type=ToolType.TOOL,
capability=ToolCapability(writes=False, read_only=True, checkpointable=False),
timeout=300,
)
instance = MockInstance(name)
runtime = ToolRuntime()
runtime.register_tool(tool, instance)
ctx = ToolExecutionContext(plan_id="m2-robot-plan")
# Discover
desc = runtime.discover(name)
if desc.name == name:
print("discover-ok")
else:
print(f"discover-fail: {desc.name}")
return 1
# Activate
runtime.activate(name, ctx)
if instance.activated:
print("activate-ok")
else:
print("activate-fail")
return 1
# Execute
result = runtime.execute(name, {"text": "hello"}, ctx)
if result.success:
print("execute-ok")
else:
print(f"execute-fail: {result.error}")
return 1
# Deactivate
runtime.deactivate(name, ctx)
if instance.deactivated:
print("deactivate-ok")
else:
print("deactivate-fail")
return 1
return 0
def _mcp_stub() -> int:
from mcp_stub_server import McpStubServer
server = McpStubServer()
server.start()
tools = server.discover()
print(f"mcp-tool-count: {len(tools)}")
result = server.invoke("mcp/search", {"query": "test"})
if "results" in result and len(result["results"]) > 0:
print("mcp-invoke-ok: mcp/search")
else:
print("mcp-invoke-fail")
return 1
server.stop()
if not server.is_running:
print("mcp-stopped")
else:
print("mcp-stop-fail")
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
+68
View File
@@ -0,0 +1,68 @@
*** Settings ***
Documentation M2 actor + tool source CLI smoke suite
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_m2_actor_tool_smoke.py
*** Test Cases ***
Actor Load Hierarchical YAML Fixture
[Documentation] Load hierarchical actor from M2 fixture YAML and verify
${result}= Run Process ${PYTHON} ${HELPER} actor-load-fixture cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-name: m2test/hierarchical-workflow
Should Contain ${result.stdout} actor-type: graph
Should Contain ${result.stdout} node-count: 3
Actor Discovery From Fixture Directory
[Documentation] Discover actors from the M2 fixtures directory
${result}= Run Process ${PYTHON} ${HELPER} actor-discover cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} actor-count: 1
Should Contain ${result.stdout} actor-loaded: m2test/hierarchical-workflow
Skill Pack Load From YAML Fixture
[Documentation] Load skill pack from M2 fixture YAML and verify
${result}= Run Process ${PYTHON} ${HELPER} skill-load-fixture cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-name: m2test/file-ops-pack
Should Contain ${result.stdout} tool-ref-count: 2
Should Contain ${result.stdout} inline-tool-count: 1
Skill Registry And Tool Resolution
[Documentation] Register skill pack and verify tool resolution
${result}= Run Process ${PYTHON} ${HELPER} skill-registry cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skill-registered: m2test/file-ops-pack
Should Contain ${result.stdout} resolved-tools-ok
Tool Lifecycle Smoke
[Documentation] Verify discover/activate/execute/deactivate lifecycle
${result}= Run Process ${PYTHON} ${HELPER} tool-lifecycle cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} discover-ok
Should Contain ${result.stdout} activate-ok
Should Contain ${result.stdout} execute-ok
Should Contain ${result.stdout} deactivate-ok
MCP Stub Discovery And Invocation
[Documentation] Start MCP stub, discover tools, invoke, and stop
${result}= Run Process ${PYTHON} ${HELPER} mcp-stub cwd=${WORKSPACE} timeout=30s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} mcp-tool-count: 3
Should Contain ${result.stdout} mcp-invoke-ok: mcp/search
Should Contain ${result.stdout} mcp-stopped