cleveragents-core/benchmarks/subplan_actor_tool_bench.py

"""ASV benchmarks for the builtin/plan-subplan actor tool.

Measures:
- SubplanPayload construction and validation
- Tool handler invocation (serial and parallel spawn)
- Decision emission with injected stub service
- Tool registration overhead
- ToolRunner round-trip execution
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

try:
    from cleveragents.domain.models.core.decision import Decision
    from cleveragents.tool.builtins import register_subplan_tool
    from cleveragents.tool.builtins.subplan_tool import (
        PLAN_SUBPLAN_SPEC,
        SubplanPayload,
        make_plan_subplan_spec,
    )
    from cleveragents.tool.registry import ToolRegistry
    from cleveragents.tool.runner import ToolRunner
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.domain.models.core.decision import Decision
    from cleveragents.tool.builtins import register_subplan_tool
    from cleveragents.tool.builtins.subplan_tool import (
        PLAN_SUBPLAN_SPEC,
        SubplanPayload,
        make_plan_subplan_spec,
    )
    from cleveragents.tool.registry import ToolRegistry
    from cleveragents.tool.runner import ToolRunner

_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"


def _serial_inputs(**extra: Any) -> dict[str, Any]:
    return {
        "goal": "Refactor authentication module",
        "plan_id": _PLAN_ID,
        "sequence_number": 0,
        "resource_scopes": ["src/auth/"],
        "parallel": False,
        **extra,
    }


def _parallel_inputs(**extra: Any) -> dict[str, Any]:
    return {
        "goal": "Run full test suite in parallel",
        "plan_id": _PLAN_ID,
        "sequence_number": 1,
        "resource_scopes": ["tests/"],
        "parallel": True,
        "max_parallel": 4,
        "dependencies": [],
        **extra,
    }


# ---------------------------------------------------------------------------
# Stub service (no DB)
# ---------------------------------------------------------------------------


class _StubService:
    def __init__(self) -> None:
        self.recorded: list[Decision] = []

    def record_decision(self, decision: Decision) -> Decision:
        self.recorded.append(decision)
        return decision


# ---------------------------------------------------------------------------
# SubplanPayload validation benchmarks
# ---------------------------------------------------------------------------


class SubplanPayloadSuite:
    """Benchmark SubplanPayload construction and model validation."""

    def time_serial_payload_construction(self) -> None:
        SubplanPayload(
            goal="Refactor auth",
            plan_id=_PLAN_ID,
            sequence_number=0,
            resource_scopes=["src/auth/"],
        )

    def time_parallel_payload_construction(self) -> None:
        SubplanPayload(
            goal="Run tests",
            plan_id=_PLAN_ID,
            sequence_number=1,
            resource_scopes=["tests/"],
            parallel=True,
            max_parallel=8,
            dependencies=["01ARZ3NDEKTSV4RRFFQ69G5FAW"],
        )

    def time_payload_with_all_fields(self) -> None:
        SubplanPayload(
            goal="Full config",
            plan_id=_PLAN_ID,
            sequence_number=2,
            resource_scopes=["src/", "tests/"],
            project_ref="projects/main",
            parallel=True,
            max_parallel=10,
            merge_strategy="fail_on_conflict",  # type: ignore[arg-type]
            dependencies=["01ARZ3NDEKTSV4RRFFQ69G5FAW"],
            context_view="executor",
        )


# ---------------------------------------------------------------------------
# Handler invocation benchmarks
# ---------------------------------------------------------------------------


class SubplanHandlerSuite:
    """Benchmark the tool handler for serial and parallel spawns."""

    def setup(self) -> None:
        self.handler = PLAN_SUBPLAN_SPEC.handler

    def time_serial_spawn(self) -> None:
        self.handler(_serial_inputs())

    def time_parallel_spawn(self) -> None:
        self.handler(_parallel_inputs())

    def time_serial_with_project_ref(self) -> None:
        self.handler(
            {
                "goal": "Migrate database schema",
                "plan_id": _PLAN_ID,
                "sequence_number": 3,
                "project_ref": "projects/db",
                "parallel": False,
            }
        )

    def time_serial_with_context_view(self) -> None:
        self.handler(_serial_inputs(context_view="executor"))


# ---------------------------------------------------------------------------
# Decision service injection benchmark
# ---------------------------------------------------------------------------


class SubplanServiceSuite:
    """Benchmark handler with injected decision service."""

    def setup(self) -> None:
        self._stub = _StubService()
        spec = make_plan_subplan_spec(decision_service=self._stub)
        self.handler = spec.handler

    def teardown(self) -> None:
        self._stub.recorded.clear()

    def time_handler_with_service(self) -> None:
        self.handler(_serial_inputs())


# ---------------------------------------------------------------------------
# Registration and ToolRunner benchmarks
# ---------------------------------------------------------------------------


class SubplanRegistrationSuite:
    """Benchmark tool registration overhead."""

    def time_register_subplan_tool(self) -> None:
        registry = ToolRegistry()
        register_subplan_tool(registry)

    def time_make_plan_subplan_spec(self) -> None:
        make_plan_subplan_spec()

    def time_make_plan_subplan_spec_with_service(self) -> None:
        stub = _StubService()
        make_plan_subplan_spec(decision_service=stub)


class SubplanRunnerSuite:
    """Benchmark ToolRunner round-trip for builtin/plan-subplan."""

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

    def time_runner_serial_spawn(self) -> None:
        self.runner.execute("builtin/plan-subplan", _serial_inputs())

    def time_runner_parallel_spawn(self) -> None:
        self.runner.execute("builtin/plan-subplan", _parallel_inputs())