cleveragents-core/benchmarks/plan_execute_bench.py

"""Airspeed Velocity benchmarks for plan execute runtime integration.

Measures construction, changeset operations, RuntimeExecuteActor
dispatch, and PlanExecutor mode selection overhead.
"""

from __future__ import annotations

from unittest.mock import MagicMock

from ulid import ULID

from cleveragents.application.services.plan_execution_context import (
    PlanExecutionContext,
    RuntimeExecuteActor,
    RuntimeExecuteResult,
)
from cleveragents.application.services.plan_executor import (
    PlanExecutor,
    StrategyDecision,
)
from cleveragents.domain.models.core.change import (
    ChangeEntry,
    ChangeOperation,
    InMemoryChangeSetStore,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner

_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
_RESOURCE_ID = "01HGZ6FE0AQDYTR4BXVQZ6EB00"


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


def _make_decisions(count: int) -> list[StrategyDecision]:
    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 _make_entry(plan_id: str, idx: int = 0) -> ChangeEntry:
    return ChangeEntry(
        plan_id=plan_id,
        resource_id=_RESOURCE_ID,
        tool_name="bench/tool",
        operation=ChangeOperation.MODIFY,
        path=f"src/bench_{idx}.py",
    )


class PlanExecutionContextSuite:
    """Benchmark PlanExecutionContext construction and operations."""

    def time_minimal_context_creation(self) -> None:
        """Time creating context with only plan_id."""
        PlanExecutionContext(plan_id=_PLAN_ID)

    def time_full_context_creation(self) -> None:
        """Time creating context with all optional fields."""
        PlanExecutionContext(
            plan_id=_PLAN_ID,
            decision_root_id=_RESOURCE_ID,
            sandbox_root="/tmp/sandbox",
            automation_profile="trusted",
            project_resources={"repo": {"path": "/code"}},
            changeset_store=InMemoryChangeSetStore(),
        )

    def time_start_changeset(self) -> None:
        """Time starting a new changeset."""
        ctx = PlanExecutionContext(plan_id=_PLAN_ID)
        ctx.start_changeset()

    def time_record_change_single(self) -> None:
        """Time recording a single change entry."""
        ctx = PlanExecutionContext(plan_id=_PLAN_ID)
        ctx.start_changeset()
        ctx.record_change(_make_entry(_PLAN_ID))

    def time_record_change_batch_10(self) -> None:
        """Time recording 10 change entries."""
        ctx = PlanExecutionContext(plan_id=_PLAN_ID)
        ctx.start_changeset()
        for i in range(10):
            ctx.record_change(_make_entry(_PLAN_ID, i))

    def time_summarize_empty(self) -> None:
        """Time summarize with no changesets."""
        ctx = PlanExecutionContext(plan_id=_PLAN_ID)
        ctx.summarize()

    def time_summarize_with_changeset(self) -> None:
        """Time summarize with one changeset and 5 entries."""
        ctx = PlanExecutionContext(plan_id=_PLAN_ID)
        ctx.start_changeset()
        for i in range(5):
            ctx.record_change(_make_entry(_PLAN_ID, i))
        ctx.summarize()


class RuntimeExecuteResultSuite:
    """Benchmark RuntimeExecuteResult model creation."""

    def time_result_creation_minimal(self) -> None:
        """Time creating result with only required fields."""
        RuntimeExecuteResult(changeset_id=str(ULID()))

    def time_result_creation_full(self) -> None:
        """Time creating result with all fields populated."""
        RuntimeExecuteResult(
            changeset_id=str(ULID()),
            tool_call_count=42,
            sandbox_refs=["/tmp/sb1", "/tmp/sb2"],
            decision_ids_processed=[str(ULID()) for _ in range(5)],
            execution_duration_ms=1500.0,
        )


class RuntimeExecuteActorSuite:
    """Benchmark RuntimeExecuteActor execution."""

    def setup(self) -> None:
        self.store = InMemoryChangeSetStore()
        self.runner = _make_runner()

    def time_single_decision_execution(self) -> None:
        """Time executing a single decision."""
        ctx = PlanExecutionContext(
            plan_id=_PLAN_ID,
            changeset_store=self.store,
        )
        actor = RuntimeExecuteActor(
            tool_runner=self.runner,
            execution_context=ctx,
        )
        actor.execute(decisions=_make_decisions(1))

    def time_five_decision_execution(self) -> None:
        """Time executing five decisions."""
        ctx = PlanExecutionContext(
            plan_id=_PLAN_ID,
            changeset_store=self.store,
        )
        actor = RuntimeExecuteActor(
            tool_runner=self.runner,
            execution_context=ctx,
        )
        actor.execute(decisions=_make_decisions(5))

    def time_execution_with_callback(self) -> None:
        """Time executing with a stream callback."""
        ctx = PlanExecutionContext(
            plan_id=_PLAN_ID,
            changeset_store=self.store,
        )
        actor = RuntimeExecuteActor(
            tool_runner=self.runner,
            execution_context=ctx,
        )
        events: list[tuple[str, dict]] = []
        actor.execute(
            decisions=_make_decisions(3),
            stream_callback=lambda t, d: events.append((t, d)),
        )


class PlanExecutorModeSuite:
    """Benchmark PlanExecutor construction with different modes."""

    def time_stub_mode_construction(self) -> None:
        """Time constructing PlanExecutor in stub mode."""
        PlanExecutor(lifecycle_service=MagicMock())

    def time_runtime_mode_construction(self) -> None:
        """Time constructing PlanExecutor in runtime mode."""
        ctx = PlanExecutionContext(plan_id=_PLAN_ID)
        PlanExecutor(
            lifecycle_service=MagicMock(),
            tool_runner=_make_runner(),
            execution_context=ctx,
        )

    def time_has_runtime_check_stub(self) -> None:
        """Time has_runtime property check in stub mode."""
        executor = PlanExecutor(lifecycle_service=MagicMock())
        _ = executor.has_runtime

    def time_has_runtime_check_runtime(self) -> None:
        """Time has_runtime property check in runtime mode."""
        ctx = PlanExecutionContext(plan_id=_PLAN_ID)
        executor = PlanExecutor(
            lifecycle_service=MagicMock(),
            execution_context=ctx,
        )
        _ = executor.has_runtime

    def time_changeset_store_access(self) -> None:
        """Time changeset_store property access in runtime mode."""
        store = InMemoryChangeSetStore()
        ctx = PlanExecutionContext(plan_id=_PLAN_ID, changeset_store=store)
        executor = PlanExecutor(
            lifecycle_service=MagicMock(),
            execution_context=ctx,
        )
        _ = executor.changeset_store