cleveragents-core/benchmarks/plan_diff_bench.py

"""ASV benchmarks for plan diff rendering overhead.

Measures the performance of:
- Diff rendering in rich format
- Diff rendering in plain format
- Diff rendering in JSON format
- Artifacts summary building
- Empty changeset guard check
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path

# 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 cleveragents.application.services.plan_apply_service import (  # noqa: E402
    PlanApplyService,
    _render_diff_json,
    _render_diff_plain,
    _render_diff_rich,
)
from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
    PlanLifecycleService,
)
from cleveragents.config.settings import Settings  # noqa: E402
from cleveragents.domain.models.core.change import (  # noqa: E402
    ChangeEntry,
    ChangeOperation,
    InMemoryChangeSetStore,
    SpecChangeSet,
)
from cleveragents.domain.models.core.plan import PlanPhase  # noqa: E402


def _build_changeset(plan_id: str, n_entries: int = 10) -> SpecChangeSet:
    """Build a SpecChangeSet with *n_entries* for benchmarking."""
    entries = []
    ops = [ChangeOperation.CREATE, ChangeOperation.MODIFY, ChangeOperation.DELETE]
    for i in range(n_entries):
        entries.append(
            ChangeEntry(
                plan_id=plan_id,
                resource_id=f"res-{i:04d}",
                tool_name="file-write",
                operation=ops[i % len(ops)],
                path=f"src/module_{i:04d}.py",
                before_hash=f"before{i:08x}" if i % 3 != 0 else None,
                after_hash=f"after{i:08x}" if i % 3 != 2 else None,
            )
        )
    return SpecChangeSet(
        changeset_id="bench-cs-001",
        plan_id=plan_id,
        entries=entries,
    )


def _create_plan_and_service() -> tuple[str, PlanApplyService, InMemoryChangeSetStore]:
    """Create a lifecycle service, plan, and apply service."""
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)
    store = InMemoryChangeSetStore()
    apply_svc = PlanApplyService(lifecycle_service=lifecycle, changeset_store=store)

    lifecycle.create_action(
        name="local/bench-diff",
        description="Benchmark diff",
        definition_of_done="Benchmarks pass",
        strategy_actor="local/planner",
        execution_actor="local/executor",
    )
    plan = lifecycle.use_action(action_name="local/bench-diff")
    plan_id = plan.identity.plan_id

    lifecycle.start_strategize(plan_id)
    lifecycle.complete_strategize(plan_id)
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.STRATEGIZE:
        lifecycle.execute_plan(plan_id)
    plan = lifecycle.get_plan(plan_id)
    if plan.phase == PlanPhase.EXECUTE:
        lifecycle.start_execute(plan_id)
        lifecycle.complete_execute(plan_id)

    cs = _build_changeset(plan_id, 20)
    store._store[cs.changeset_id] = cs
    plan = lifecycle.get_plan(plan_id)
    plan.changeset_id = cs.changeset_id
    plan.sandbox_refs = ["sandbox-bench-001"]
    lifecycle._commit_plan(plan)

    return plan_id, apply_svc, store


class DiffRenderingSuite:
    """Benchmark suite for diff rendering in various formats."""

    def setup(self) -> None:
        self.plan_id, self.apply_svc, self.store = _create_plan_and_service()

    def time_diff_rich(self) -> None:
        self.apply_svc.diff(self.plan_id, fmt="rich")

    def time_diff_plain(self) -> None:
        self.apply_svc.diff(self.plan_id, fmt="plain")

    def time_diff_json(self) -> None:
        self.apply_svc.diff(self.plan_id, fmt="json")

    def time_diff_yaml(self) -> None:
        self.apply_svc.diff(self.plan_id, fmt="yaml")


class ArtifactsSuite:
    """Benchmark suite for artifacts rendering."""

    def setup(self) -> None:
        self.plan_id, self.apply_svc, self.store = _create_plan_and_service()

    def time_artifacts_json(self) -> None:
        self.apply_svc.artifacts(self.plan_id, fmt="json")

    def time_artifacts_plain(self) -> None:
        self.apply_svc.artifacts(self.plan_id, fmt="plain")


class GuardSuite:
    """Benchmark suite for empty changeset guard."""

    def setup(self) -> None:
        self.plan_id, self.apply_svc, self.store = _create_plan_and_service()

    def time_guard_non_empty(self) -> None:
        self.apply_svc.guard_empty_changeset(self.plan_id)


class HelperFunctionsSuite:
    """Benchmark suite for low-level rendering helpers."""

    def setup(self) -> None:
        self.changeset = _build_changeset("bench-plan-001", 50)

    def time_render_diff_plain(self) -> None:
        _render_diff_plain(self.changeset)

    def time_render_diff_rich(self) -> None:
        _render_diff_rich(self.changeset)

    def time_render_diff_json(self) -> None:
        _render_diff_json(self.changeset)