cleveragents-core/benchmarks/execution_throughput_bench.py

"""ASV benchmarks for plan execution throughput at scale.

Measures sequential and concurrent plan execution overhead at varying
plan counts (10, 50, 100).  Uses the lightweight in-process executor
path (no database, no LLM) to isolate execution-dispatch cost.
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path
from typing import ClassVar
from unittest.mock import MagicMock

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

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from ulid import ULID  # noqa: E402

from cleveragents.application.services.plan_execution_context import (  # noqa: E402
    PlanExecutionContext,
    RuntimeExecuteActor,
)
from cleveragents.application.services.plan_executor import (  # noqa: E402
    PlanExecutor,
    StrategyDecision,
)
from cleveragents.domain.models.core.change import (  # noqa: E402
    InMemoryChangeSetStore,
)
from cleveragents.tool.registry import ToolRegistry  # noqa: E402
from cleveragents.tool.runner import ToolRunner  # noqa: E402

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

_RESOURCE_ID = "01HGZ6FE0AQDYTR4BXVQZ6EB00"


def _make_runner() -> ToolRunner:
    return ToolRunner(registry=ToolRegistry())


def _make_decisions(count: int) -> list[StrategyDecision]:
    """Build a linear chain of *count* decisions."""
    root_id = str(ULID())
    return [
        StrategyDecision(
            decision_id=root_id if i == 0 else str(ULID()),
            step_text=f"Step {i + 1}",
            sequence=i,
            parent_id=root_id if i > 0 else None,
        )
        for i in range(count)
    ]


def _execute_single_plan(runner: ToolRunner) -> None:
    """Execute one plan with 3 decisions (fire-and-forget)."""
    plan_id = str(ULID())
    ctx = PlanExecutionContext(
        plan_id=plan_id,
        changeset_store=InMemoryChangeSetStore(),
    )
    actor = RuntimeExecuteActor(tool_runner=runner, execution_context=ctx)
    actor.execute(decisions=_make_decisions(3))


# ---------------------------------------------------------------------------
# Parameterized execution throughput suite
# ---------------------------------------------------------------------------


class ExecutionThroughputSuite:
    """Benchmark plan execution throughput at varying plan counts."""

    params: ClassVar[list[int]] = [10, 50, 100]
    param_names: ClassVar[list[str]] = ["plan_count"]
    timeout = 300

    _runner: ToolRunner

    def setup(self, plan_count: int) -> None:
        """Prepare a shared tool runner."""
        self._runner = _make_runner()

    def time_sequential_plans(self, plan_count: int) -> None:
        """Execute *plan_count* plans sequentially."""
        for _ in range(plan_count):
            _execute_single_plan(self._runner)

    def time_executor_construction(self, plan_count: int) -> None:
        """Construct *plan_count* PlanExecutor instances."""
        lifecycle = MagicMock()
        for _ in range(plan_count):
            ctx = PlanExecutionContext(
                plan_id=str(ULID()),
                changeset_store=InMemoryChangeSetStore(),
            )
            PlanExecutor(
                lifecycle_service=lifecycle,
                tool_runner=self._runner,
                execution_context=ctx,
            )

    def time_decision_tree_scaling(self, plan_count: int) -> None:
        """Execute one plan with *plan_count* decisions."""
        plan_id = str(ULID())
        ctx = PlanExecutionContext(
            plan_id=plan_id,
            changeset_store=InMemoryChangeSetStore(),
        )
        actor = RuntimeExecuteActor(tool_runner=self._runner, execution_context=ctx)
        actor.execute(decisions=_make_decisions(plan_count))