cleveragents-core/robot/helper_actor_hierarchy.py

"""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",
        "provider": "openai",
        "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",
        "provider": "openai",
        "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",
        "provider": "openai",
        "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",
        "provider": "openai",
        "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())