cleveragents-core/benchmarks/tool_lifecycle_bench.py

"""ASV benchmarks for tool lifecycle runtime."""

from __future__ import annotations

from typing import Any

from cleveragents.tool.context import (
    BoundResource,
    CancellationToken,
    Change,
    ChangeOperation,
    ToolExecutionContext,
)
from cleveragents.tool.lifecycle import (
    ToolDescriptor,
    ToolLifecycleCache,
    ToolResult,
)
from cleveragents.tool.schema_validator import validate_tool_input


class _DummyInstance:
    """Minimal mock for benchmarks."""

    def discover(self) -> ToolDescriptor:
        return ToolDescriptor(name="bench/tool", description="bench", source="builtin")

    def activate(self, ctx: Any) -> None:
        pass

    def execute(self, params: Any, ctx: Any) -> ToolResult:
        return ToolResult()

    def deactivate(self, ctx: Any) -> None:
        pass


class TimeToolExecutionContext:
    """Benchmark ToolExecutionContext creation and operations."""

    def time_create_context(self) -> None:
        ToolExecutionContext(plan_id="plan-bench-001")

    def time_create_context_with_resources(self) -> None:
        resources = {
            f"slot_{i}": BoundResource(
                slot_name=f"slot_{i}",
                resource_id=f"res-{i:03d}",
                resource_type="git-checkout",
                access="read_write",
            )
            for i in range(10)
        }
        ToolExecutionContext(
            plan_id="plan-bench-002",
            resources=resources,
        )

    def time_record_change(self) -> None:
        ctx = ToolExecutionContext(plan_id="plan-bench-003")
        for i in range(100):
            ctx.record_change(
                Change(
                    operation=ChangeOperation.MODIFY,
                    resource_id=f"res-{i:03d}",
                )
            )

    def time_context_summary(self) -> None:
        ctx = ToolExecutionContext(plan_id="plan-bench-004")
        for _ in range(50):
            ctx.as_summary()


class TimeCancellationToken:
    """Benchmark CancellationToken operations."""

    def time_create_token(self) -> None:
        CancellationToken()

    def time_check_not_cancelled(self) -> None:
        token = CancellationToken()
        for _ in range(1000):
            _ = token.is_cancelled

    def time_cancel_and_check(self) -> None:
        token = CancellationToken()
        token.cancel()
        for _ in range(1000):
            _ = token.is_cancelled


class TimeToolLifecycleCache:
    """Benchmark ToolLifecycleCache operations."""

    def time_put_and_get(self) -> None:
        cache = ToolLifecycleCache()
        dummy = _DummyInstance()
        for i in range(100):
            cache.put(f"plan-{i}", f"tool/t-{i}", dummy)
        for i in range(100):
            cache.get(f"plan-{i}", f"tool/t-{i}")

    def time_remove_plan(self) -> None:
        cache = ToolLifecycleCache()
        dummy = _DummyInstance()
        for i in range(50):
            cache.put("plan-001", f"tool/t-{i}", dummy)
        cache.remove_plan("plan-001")

    def time_get_stats(self) -> None:
        cache = ToolLifecycleCache()
        dummy = _DummyInstance()
        for i in range(20):
            cache.put("plan-001", f"tool/t-{i}", dummy)
        for _ in range(50):
            cache.get_stats("plan-001")


class TimeSchemaValidation:
    """Benchmark JSON Schema validation."""

    def setup(self) -> None:
        self.schema: dict[str, Any] = {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"},
                "mode": {"type": "integer"},
            },
            "required": ["path"],
        }

    def time_validate_simple(self) -> None:
        validate_tool_input({"path": "test.txt"}, self.schema)

    def time_validate_full(self) -> None:
        validate_tool_input(
            {"path": "test.txt", "content": "hello", "mode": 644},
            self.schema,
        )


class TimeToolDescriptor:
    """Benchmark ToolDescriptor creation."""

    def time_create_descriptor(self) -> None:
        for i in range(100):
            ToolDescriptor(
                name=f"ns/tool-{i}",
                description=f"Tool {i}",
            )

    def time_create_descriptor_with_schema(self) -> None:
        schema: dict[str, Any] = {
            "type": "object",
            "properties": {"path": {"type": "string"}},
        }
        for i in range(100):
            ToolDescriptor(
                name=f"ns/tool-{i}",
                description=f"Tool {i}",
                input_schema=schema,
                output_schema=schema,
            )