cleveragents-core/benchmarks/actor_runtime_bench.py

"""ASV benchmarks for tool-calling actor runtime overhead.

Measures the performance of:
- ToolCallingRuntime construction
- Tool schema export
- Single-iteration tool-call loop (1 tool call)
- Multi-iteration tool-call loop (3 tool calls)
- Max-iteration termination overhead
- ToolActorContext construction and recording
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

try:
    from cleveragents.tool.actor_context import ToolActorContext, ToolCallRecord
    from cleveragents.tool.actor_runtime import (
        LLMResponse,
        LLMToolCall,
        ToolCallingRuntime,
    )
    from cleveragents.tool.registry import ToolRegistry
    from cleveragents.tool.runner import ToolRunner
    from cleveragents.tool.runtime import ToolSpec
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.tool.actor_context import ToolActorContext, ToolCallRecord
    from cleveragents.tool.actor_runtime import (
        LLMResponse,
        LLMToolCall,
        ToolCallingRuntime,
    )
    from cleveragents.tool.registry import ToolRegistry
    from cleveragents.tool.runner import ToolRunner
    from cleveragents.tool.runtime import ToolSpec


def _echo(inputs: dict[str, Any]) -> dict[str, Any]:
    return {"echoed": inputs}


class _SingleToolLLM:
    def __init__(self) -> None:
        self._n = 0

    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        self._n += 1
        if self._n == 1:
            return LLMResponse(
                tool_calls=[LLMToolCall(name="bench/echo", arguments={"x": 1})]
            )
        return LLMResponse(content="done")


class _AlwaysToolLLM:
    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        return LLMResponse(
            tool_calls=[LLMToolCall(name="bench/echo", arguments={"x": 1})]
        )


class _ImmediateLLM:
    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        return LLMResponse(content="immediate")


def _make_registry() -> ToolRegistry:
    reg = ToolRegistry()
    reg.register(ToolSpec(name="bench/echo", description="Bench echo", handler=_echo))
    return reg


class ToolCallingRuntimeConstructionSuite:
    """Benchmark ToolCallingRuntime construction."""

    def setup(self) -> None:
        self.registry = _make_registry()
        self.runner = ToolRunner(self.registry)
        self.llm = _ImmediateLLM()

    def time_construction(self) -> None:
        ToolCallingRuntime(
            registry=self.registry,
            runner=self.runner,
            llm_caller=self.llm,
        )


class ToolSchemaExportSuite:
    """Benchmark tool schema export."""

    def setup(self) -> None:
        self.registry = _make_registry()
        self.runner = ToolRunner(self.registry)
        self.llm = _ImmediateLLM()
        self.runtime = ToolCallingRuntime(
            registry=self.registry,
            runner=self.runner,
            llm_caller=self.llm,
        )

    def time_export_schemas(self) -> None:
        self.runtime.export_tool_schemas()


class SingleToolCallLoopSuite:
    """Benchmark single-iteration tool-call loop."""

    def setup(self) -> None:
        self.registry = _make_registry()
        self.runner = ToolRunner(self.registry)

    def time_single_tool_call(self) -> None:
        llm = _SingleToolLLM()
        runtime = ToolCallingRuntime(
            registry=self.registry,
            runner=self.runner,
            llm_caller=llm,
        )
        runtime.run_tool_loop(prompt="bench")


class MaxIterationSuite:
    """Benchmark max-iteration termination overhead."""

    def setup(self) -> None:
        self.registry = _make_registry()
        self.runner = ToolRunner(self.registry)

    def time_max_iterations_5(self) -> None:
        runtime = ToolCallingRuntime(
            registry=self.registry,
            runner=self.runner,
            llm_caller=_AlwaysToolLLM(),
            max_iterations=5,
        )
        runtime.run_tool_loop(prompt="bench")


class ToolActorContextSuite:
    """Benchmark ToolActorContext operations."""

    def time_context_construction(self) -> None:
        ToolActorContext(plan_id="bench-plan", phase="execute")

    def time_record_tool_call(self) -> None:
        ctx = ToolActorContext(plan_id="bench-plan")
        record = ToolCallRecord(
            tool_name="bench/tool",
            inputs={"x": 1},
            output={"y": 2},
            duration_ms=1.0,
        )
        ctx.record_tool_call(record)

    def time_context_summary(self) -> None:
        ctx = ToolActorContext(
            plan_id="bench-plan",
            phase="execute",
            sandbox_root="/tmp/sb",
        )
        ctx.as_summary()