Files
cleveragents-core/robot/helper_actor_hierarchy.py
aditya e476d2de0e feat(actor): extend hierarchical actor YAML schema and loader
Extend actor YAML schema to support hierarchical graphs with explicit
node types (agent, tool, conditional, subgraph), per-node LSP bindings
(lsp_binding with server, languages, auto, capabilities), and tool-source
references (skills, mcp_servers, agent_skills).

Add schema validation for namespaced actor references, duplicate node IDs,
edge target existence, and graph reachability — all nodes must be reachable
from entry_node via explicit edges or conditional node routing targets.

Update loader to report YAML parse errors with precise line/column positions
and schema validation errors with dotted field paths and remediation hints
pointing to docs/reference/actor_config.md.

Add docs/reference/actor_config.md as the practical configuration reference
covering hierarchical graph examples, node type table, topology rules, and
common error cases with fix guidance.

Refresh examples/actors/graph_workflow.yaml to replace deprecated actor_path
with actor_ref. Add benchmarks/actor_yaml_bench.py for schema load overhead.

Tests: 95 Behave scenarios, 10 Robot smoke tests (including hierarchical
loader smoke test), security scan clean, coverage 99% (threshold 97%).

ISSUES CLOSED: #157
2026-02-25 10:30:42 +00:00

194 lines
5.8 KiB
Python

"""Robot Framework helper for hierarchical actor YAML smoke tests.
Usage:
python robot/helper_actor_hierarchy.py discover-hierarchical
python robot/helper_actor_hierarchy.py parse-lsp-binding
python robot/helper_actor_hierarchy.py parse-tool-sources
python robot/helper_actor_hierarchy.py parse-subgraph-ref
python robot/helper_actor_hierarchy.py reject-bad-lsp
"""
from __future__ import annotations
import sys
from pathlib import Path
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.actor.loader import ActorLoader # noqa: E402
from cleveragents.actor.schema import ActorConfigSchema # noqa: E402
def main() -> int:
if len(sys.argv) < 2:
print("Usage: helper_actor_hierarchy.py <command>")
return 1
cmd = sys.argv[1]
dispatch = {
"discover-hierarchical": _discover_hierarchical,
"parse-lsp-binding": _parse_lsp_binding,
"parse-tool-sources": _parse_tool_sources,
"parse-subgraph-ref": _parse_subgraph_ref,
"reject-bad-lsp": _reject_bad_lsp,
}
handler = dispatch.get(cmd)
if handler is None:
print(f"Unknown command: {cmd}")
return 1
return handler()
def _discover_hierarchical() -> int:
project_root = Path(__file__).resolve().parents[1]
examples_dir = project_root / "examples" / "actors"
loader = ActorLoader(search_roots=[examples_dir])
actors = loader.discover()
for actor in actors:
if actor.name == "workflows/dev-workflow":
print(f"actor-loaded: {actor.name}")
print(f"skills-count: {len(actor.skills)}")
print(f"lsp-configured: {actor.lsp is not None}")
return 0
print("actor-not-found: workflows/dev-workflow")
return 1
def _parse_lsp_binding() -> int:
data = {
"name": "local/lsp-test",
"type": "graph",
"description": "LSP binding test",
"model": "gpt-4",
"route": {
"nodes": [
{
"id": "coder",
"type": "agent",
"name": "Coder",
"description": "Codes with LSP",
"config": {"model": "gpt-4"},
"lsp_binding": {
"server": "local/pyright",
"languages": ["python"],
},
},
],
"edges": [],
"entry_node": "coder",
"exit_nodes": ["coder"],
},
}
config = ActorConfigSchema.model_validate(data)
node = config.route.nodes[0] # type: ignore[union-attr]
if node.lsp_binding:
print(f"lsp-binding-server: {node.lsp_binding.server}")
print(f"lsp-binding-langs: {', '.join(node.lsp_binding.languages)}")
return 0
def _parse_tool_sources() -> int:
data = {
"name": "local/ts-test",
"type": "graph",
"description": "Tool sources test",
"model": "gpt-4",
"route": {
"nodes": [
{
"id": "exec",
"type": "agent",
"name": "Executor",
"description": "Executor with tool sources",
"config": {},
"tool_sources": [
{"type": "skill", "name": "local/file-ops"},
{"type": "mcp", "name": "local/filesystem"},
],
},
],
"edges": [],
"entry_node": "exec",
"exit_nodes": ["exec"],
},
}
config = ActorConfigSchema.model_validate(data)
node = config.route.nodes[0] # type: ignore[union-attr]
print(f"tool-sources-count: {len(node.tool_sources)}")
return 0
def _parse_subgraph_ref() -> int:
data = {
"name": "local/sg-test",
"type": "graph",
"description": "Subgraph ref test",
"model": "gpt-4",
"route": {
"nodes": [
{
"id": "main",
"type": "agent",
"name": "Main",
"description": "Main agent",
"config": {},
},
{
"id": "review",
"type": "subgraph",
"name": "Review",
"description": "Subgraph review",
"actor_ref": "local/code-reviewer",
},
],
"edges": [{"from_node": "main", "to_node": "review"}],
"entry_node": "main",
"exit_nodes": ["review"],
},
}
config = ActorConfigSchema.model_validate(data)
node = config.route.nodes[1] # type: ignore[union-attr]
print(f"actor-ref: {node.actor_ref}")
return 0
def _reject_bad_lsp() -> int:
data = {
"name": "local/bad-lsp-test",
"type": "graph",
"description": "Bad LSP test",
"model": "gpt-4",
"route": {
"nodes": [
{
"id": "n",
"type": "agent",
"name": "N",
"description": "Bad LSP",
"config": {},
"lsp_binding": {"server": "no-slash"},
},
],
"edges": [],
"entry_node": "n",
"exit_nodes": ["n"],
},
}
try:
ActorConfigSchema.model_validate(data)
print("validation-unexpected-success")
return 1
except Exception as exc:
err = str(exc).lower()
if "lsp_binding" in err:
print("validation-error: lsp_binding")
else:
print(f"validation-error-other: {exc}")
return 0
if __name__ == "__main__":
sys.exit(main())