cleveragents-core/benchmarks/changeset_persistence_bench.py

"""ASV benchmarks for ChangeSet persistence overhead.

Measures the cost of persisting ChangeEntry records and retrieving
them via the SqliteChangeSetStore, simulating typical plan execution
workloads.
"""

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

from cleveragents.domain.models.core.change import (
    ChangeEntry,
    ChangeOperation,
)
from cleveragents.infrastructure.database.changeset_repository import (
    ChangeSetEntryRepository,
    SqliteChangeSetStore,
)
from cleveragents.infrastructure.database.models import Base


def _make_factory() -> tuple[sessionmaker[Session], object]:
    """Create an in-memory SQLite engine + session factory."""
    engine = create_engine(
        "sqlite:///:memory:",
        echo=False,
        connect_args={"check_same_thread": False},
    )
    Base.metadata.create_all(engine)
    factory: sessionmaker[Session] = sessionmaker(bind=engine)
    return factory, engine


def _make_entry(plan_id: str, idx: int) -> ChangeEntry:
    return ChangeEntry(
        plan_id=plan_id,
        resource_id="res-001",
        tool_name="file-write",
        operation=ChangeOperation.CREATE,
        path=f"src/file_{idx}.py",
        after_hash=f"hash_{idx:06d}",
    )


class TimeChangeSetPersistence:
    """Benchmark suite for ChangeSet persistence operations."""

    timeout = 60.0

    def setup(self) -> None:
        self.factory, self.engine = _make_factory()
        self.store = SqliteChangeSetStore(self.factory)
        self.plan_id = "bench-plan-001"
        self.changeset_id = self.store.start(self.plan_id)
        # Pre-populate for retrieval benchmarks
        for i in range(50):
            entry = _make_entry(self.plan_id, i)
            self.store.record(self.changeset_id, entry)
        session = self.factory()
        session.commit()
        session.close()

    def time_record_single_entry(self) -> None:
        """Time recording a single ChangeEntry."""
        entry = _make_entry(self.plan_id, 999)
        self.store.record(self.changeset_id, entry)

    def time_get_changeset_50_entries(self) -> None:
        """Time retrieving a changeset with 50 entries."""
        self.store.get(self.changeset_id)

    def time_summarize_changeset(self) -> None:
        """Time computing summary for a changeset."""
        self.store.summarize(self.changeset_id)

    def time_start_changeset(self) -> None:
        """Time starting a new empty changeset."""
        self.store.start("bench-plan-new")


class TimeEntryRepositoryBulk:
    """Benchmark bulk entry operations."""

    timeout = 60.0

    def setup(self) -> None:
        self.factory, self.engine = _make_factory()
        self.repo = ChangeSetEntryRepository(self.factory)

    def time_save_100_entries(self) -> None:
        """Time saving 100 entries sequentially."""
        for i in range(100):
            entry = _make_entry("bench-bulk", i)
            self.repo.save_entry(f"cs-bulk-{i % 5}", entry)

    def time_retrieve_by_plan(self) -> None:
        """Time retrieving all entries for a plan."""
        self.repo.get_entries_for_plan("bench-bulk")