temp/benchmarks/m2_actor_tool_smoke_bench.py

"""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
from pathlib import Path
from typing import Any

# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

# Make the MCP stub mock importable (lives in features/mocks/, not installed).
_MOCKS = str(Path(__file__).resolve().parents[1] / "features" / "mocks")
if _MOCKS not in sys.path:
    sys.path.insert(0, _MOCKS)

# Force-reload the top-level package so Python picks up the source tree
# version instead of a potentially stale installed copy.
import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from mcp_stub_server import McpStubServer  # noqa: E402

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,
)

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