cleveragents-core/benchmarks/persistence_suites_bench.py

"""ASV benchmarks for persistence test runtime baseline.

Measures action and plan repository operation latencies to track
performance regressions in persistence operations.
"""

from __future__ import annotations

from datetime import datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from cleveragents.domain.models.core.action import (
    Action,
    ActionArgument,
    ActionState,
    ArgumentRequirement,
    ArgumentType,
)
from cleveragents.domain.models.core.plan import (
    NamespacedName,
    PlanIdentity,
    PlanPhase,
    PlanTimestamps,
    ProcessingState,
)
from cleveragents.domain.models.core.plan import Plan as V3Plan
from cleveragents.infrastructure.database.models import Base
from cleveragents.infrastructure.database.repositories import (
    ActionRepository,
    LifecyclePlanRepository,
)

_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
_bench_counter = 0


def _bench_ulid() -> str:
    global _bench_counter
    _bench_counter += 1
    n = _bench_counter
    suffix = ""
    for _ in range(8):
        suffix = _CB32[n % 32] + suffix
        n //= 32
    return f"01HGZPS10AQDYTR4PS{suffix}"


def _make_bench_action(name: str = "local/ps-bench-action") -> Action:
    """Create a benchmark action with arguments and invariants."""
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="Persistence suite benchmark action",
        long_description=None,
        definition_of_done="Done",
        strategy_actor="local/s",
        execution_actor="local/e",
        estimation_actor=None,
        review_actor=None,
        arguments=[
            ActionArgument(
                name="target",
                arg_type=ArgumentType.STRING,
                requirement=ArgumentRequirement.REQUIRED,
                description="Target path",
            ),
        ],
        invariants=["No breaking changes"],
        reusable=True,
        read_only=False,
        state=ActionState.AVAILABLE,
        created_at=datetime.now(),
        updated_at=datetime.now(),
        created_by=None,
        tags=[],
    )


def _make_bench_plan(plan_id: str | None = None) -> V3Plan:
    now = datetime.now()
    return V3Plan(
        identity=PlanIdentity(plan_id=plan_id or _bench_ulid(), attempt=1),
        namespaced_name=NamespacedName(namespace="local", name="ps-bench-plan"),
        action_name="local/ps-bench-action",
        description="Persistence suite benchmark plan",
        phase=PlanPhase("action"),
        processing_state=ProcessingState("queued"),
        strategy_actor="local/s",
        execution_actor="local/e",
        timestamps=PlanTimestamps(created_at=now, updated_at=now),
        created_by="bench",
        tags=[],
        reusable=True,
        read_only=False,
    )


class TimeActionCreate:
    """Benchmark action creation latency."""

    timeout = 60

    def setup(self) -> None:
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()
        self.session = session
        self.repo = ActionRepository(session_factory=lambda: session)
        self._counter = 0

    def time_create_action(self) -> None:
        self._counter += 1
        action = _make_bench_action(f"local/bench-act-{self._counter}")
        self.repo.create(action)
        self.session.commit()


class TimeActionGet:
    """Benchmark action retrieval latency."""

    timeout = 60

    def setup(self) -> None:
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()
        self.session = session
        self.repo = ActionRepository(session_factory=lambda: session)
        action = _make_bench_action()
        self.repo.create(action)
        session.commit()

    def time_get_by_name(self) -> None:
        self.repo.get_by_name("local/ps-bench-action")


class TimePlanCreatePersistence:
    """Benchmark plan creation with FK action."""

    timeout = 60

    def setup(self) -> None:
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()
        self.session = session
        factory = lambda: session  # noqa: E731
        action_repo = ActionRepository(session_factory=factory)
        action_repo.create(_make_bench_action())
        session.commit()
        self.repo = LifecyclePlanRepository(session_factory=factory)

    def time_create_plan(self) -> None:
        plan = _make_bench_plan()
        self.repo.create(plan)
        self.session.commit()


class TimePlanUpdatePhase:
    """Benchmark plan phase update latency."""

    timeout = 60

    def setup(self) -> None:
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()
        self.session = session
        factory = lambda: session  # noqa: E731
        action_repo = ActionRepository(session_factory=factory)
        action_repo.create(_make_bench_action())
        session.commit()
        self.repo = LifecyclePlanRepository(session_factory=factory)
        self.plan_id = _bench_ulid()
        plan = _make_bench_plan(self.plan_id)
        self.repo.create(plan)
        session.commit()

    def time_update_phase(self) -> None:
        plan = self.repo.get(self.plan_id)
        if plan is not None:
            plan.phase = PlanPhase.STRATEGIZE
            plan.processing_state = ProcessingState.PROCESSING
            plan.timestamps.updated_at = datetime.now()
            self.repo.update(plan)
            self.session.commit()


class TimePlanListFiltered:
    """Benchmark plan listing with filters."""

    timeout = 60

    def setup(self) -> None:
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()
        self.session = session
        factory = lambda: session  # noqa: E731
        action_repo = ActionRepository(session_factory=factory)
        action_repo.create(_make_bench_action())
        session.commit()
        self.repo = LifecyclePlanRepository(session_factory=factory)
        for _ in range(30):
            plan = _make_bench_plan()
            self.repo.create(plan)
        session.commit()

    def time_list_all(self) -> None:
        self.repo.list_plans()

    def time_list_by_phase(self) -> None:
        self.repo.list_plans(phase="action")

    def time_count(self) -> None:
        self.repo.count()