temp/robot/helper_m2_actor_tool_smoke.py

"""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())