cleveragents-core/benchmarks/persistence_robot_bench.py

"""ASV benchmarks for Robot persistence fixture setup overhead.

Measures the cost of initialising file-backed SQLite databases,
seeding prerequisite actions, creating plans, and performing the
close-reopen cycle used in Robot persistence lifecycle tests.
"""

from __future__ import annotations

import tempfile
from datetime import UTC, datetime
from pathlib import Path

from sqlalchemy import create_engine
from sqlalchemy.orm import Session

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

_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
_bench_counter = 9000


def _bench_ulid() -> str:
    """Generate a monotonically-increasing ULID for benchmarks."""
    global _bench_counter
    _bench_counter += 1
    n: int = _bench_counter
    suffix: str = ""
    for _ in range(8):
        suffix = _CB32[n % 32] + suffix
        n //= 32
    return f"01HGRBT10AQDYTR4RB{suffix}"


def _make_bench_action(name: str = "local/rb-bench-action") -> Action:
    """Create a benchmark action with arguments and invariants."""
    now: datetime = datetime(2026, 2, 1, tzinfo=UTC)
    return Action(
        namespaced_name=NamespacedName.parse(name),
        description="Robot bench action",
        definition_of_done="Bench done",
        strategy_actor="local/strat",
        execution_actor="local/exec",
        arguments=[
            ActionArgument(
                name="target",
                arg_type=ArgumentType.STRING,
                requirement=ArgumentRequirement.REQUIRED,
                description="Target path",
            ),
        ],
        invariants=["No regressions"],
        state=ActionState.AVAILABLE,
        created_at=now,
        updated_at=now,
    )


def _make_bench_plan(
    plan_id: str | None = None,
    action_name: str = "local/rb-bench-action",
) -> Plan:
    """Create a benchmark plan."""
    now: datetime = datetime(2026, 3, 1, tzinfo=UTC)
    return Plan(
        identity=PlanIdentity(plan_id=plan_id or _bench_ulid(), attempt=1),
        namespaced_name=NamespacedName(namespace="local", name="rb-bench-plan"),
        action_name=action_name,
        description="Robot bench plan",
        phase=PlanPhase.ACTION,
        processing_state=ProcessingState.QUEUED,
        automation_profile=AutomationProfileRef(
            profile_name="trusted",
            provenance=AutomationProfileProvenance.PLAN,
        ),
        strategy_actor="local/strat",
        execution_actor="local/exec",
        project_links=[ProjectLink(project_name="local/proj-bench")],
        timestamps=PlanTimestamps(created_at=now, updated_at=now),
        created_by="bench",
        tags=[],
        reusable=True,
        read_only=False,
    )


class TimeRobotFixtureSetupSuite:
    """Benchmark Robot fixture initialisation overhead."""

    timeout = 60

    def setup(self) -> None:
        """Pre-allocate nothing -- each time_* creates its own DB."""

    def time_file_db_init(self) -> None:
        """Measure file-backed SQLite init + schema creation."""
        tmp: str = tempfile.mktemp(suffix=".db")
        try:
            init_database(f"sqlite:///{tmp}")
        finally:
            Path(tmp).unlink(missing_ok=True)

    def time_seed_action(self) -> None:
        """Measure action seeding into a fresh file DB."""
        tmp: str = tempfile.mktemp(suffix=".db")
        try:
            engine = init_database(f"sqlite:///{tmp}")
            session = Session(bind=engine)
            factory = lambda: session  # noqa: E731
            repo = ActionRepository(session_factory=factory)
            repo.create(_make_bench_action())
            session.commit()
            session.close()
        finally:
            Path(tmp).unlink(missing_ok=True)

    def time_seed_plan(self) -> None:
        """Measure plan creation including FK action seeding."""
        tmp: str = tempfile.mktemp(suffix=".db")
        try:
            engine = init_database(f"sqlite:///{tmp}")
            session = Session(bind=engine)
            factory = lambda: session  # noqa: E731
            action_repo = ActionRepository(session_factory=factory)
            action_repo.create(_make_bench_action())
            session.commit()
            plan_repo = LifecyclePlanRepository(session_factory=factory)
            plan_repo.create(_make_bench_plan())
            session.commit()
            session.close()
        finally:
            Path(tmp).unlink(missing_ok=True)


class TimeRobotRestartCycleSuite:
    """Benchmark the close-reopen persistence cycle used in Robot tests."""

    timeout = 60

    def setup(self) -> None:
        """Create file DB with action + plan for restart benchmarks."""
        self._tmp: str = tempfile.mktemp(suffix=".db")
        engine = init_database(f"sqlite:///{self._tmp}")
        session = Session(bind=engine)
        factory = lambda: session  # noqa: E731
        action_repo = ActionRepository(session_factory=factory)
        action_repo.create(_make_bench_action())
        session.commit()
        plan_repo = LifecyclePlanRepository(session_factory=factory)
        self._plan_id: str = _bench_ulid()
        plan_repo.create(_make_bench_plan(self._plan_id))
        session.commit()
        session.close()
        engine.dispose()

    def time_close_reopen_verify(self) -> None:
        """Measure close + reopen + get cycle latency."""
        engine2 = create_engine(f"sqlite:///{self._tmp}", echo=False)
        session2 = Session(bind=engine2)
        factory2 = lambda: session2  # noqa: E731
        repo2 = LifecyclePlanRepository(session_factory=factory2)
        plan = repo2.get(self._plan_id)
        assert plan is not None
        session2.close()
        engine2.dispose()

    def teardown(self) -> None:
        """Clean up temp DB file."""
        Path(self._tmp).unlink(missing_ok=True)