cleveragents-core/benchmarks/resource_registry_bench.py

"""ASV benchmarks for resource registry lookup operations.

Measures the performance of:
- ResourceTypeSpec construction from config dict
- ResourceRegistryService.show_type() lookup
- ResourceRegistryService.show_resource() lookup by name and ULID
- ResourceRegistryService.list_types() enumeration
- ResourceRegistryService.list_resources() enumeration
- ResourceRegistryService.register_resource() creation
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

try:
    from cleveragents.application.services.resource_registry_service import (
        ResourceRegistryService,
    )
    from cleveragents.domain.models.core.resource_type import (
        ResourceTypeSpec,
    )
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.application.services.resource_registry_service import (
        ResourceRegistryService,
    )
    from cleveragents.domain.models.core.resource_type import (
        ResourceTypeSpec,
    )


def _setup_db() -> Any:
    """Create in-memory database and return session factory."""
    from sqlalchemy import create_engine, event
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool

    from cleveragents.infrastructure.database.models import Base

    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

    @event.listens_for(engine, "connect")
    def _fk(conn: Any, _rec: Any) -> None:
        conn.cursor().execute("PRAGMA foreign_keys=ON")

    Base.metadata.create_all(engine)
    return sessionmaker(bind=engine)


class TypeSpecConstructionSuite:
    """Benchmark ResourceTypeSpec construction from config dicts."""

    def time_from_config_minimal(self) -> None:
        """Benchmark minimal type spec construction."""
        ResourceTypeSpec.from_config(
            {
                "name": "bench/minimal",
                "description": "Minimal benchmark type",
                "resource_kind": "physical",
                "sandbox_strategy": "copy_on_write",
            }
        )

    def time_from_config_full(self) -> None:
        """Benchmark full type spec construction with all fields."""
        ResourceTypeSpec.from_config(
            {
                "name": "bench/full",
                "description": "Full benchmark type",
                "resource_kind": "physical",
                "sandbox_strategy": "git_worktree",
                "user_addable": True,
                "built_in": False,
                "cli_args": [
                    {
                        "name": "path",
                        "type": "path",
                        "required": True,
                        "description": "Path to resource",
                    },
                    {
                        "name": "branch",
                        "type": "string",
                        "required": False,
                        "description": "Branch name",
                        "default": "main",
                    },
                ],
                "parent_types": ["git-checkout"],
                "child_types": ["fs-directory", "fs-file"],
                "handler": "bench.handler:BenchHandler",
                "capabilities": {
                    "read": True,
                    "write": True,
                    "sandbox": True,
                    "checkpoint": False,
                },
                "auto_discovery": {
                    "enabled": True,
                    "rules": [
                        {"type": "fs-directory", "pattern": "*/"},
                    ],
                },
            }
        )


class RegistryLookupSuite:
    """Benchmark registry service lookup operations."""

    def setup(self) -> None:
        self._sf = _setup_db()
        self._svc = ResourceRegistryService(session_factory=self._sf)
        self._svc.bootstrap_builtin_types()

        # Register some resources for lookup benchmarks
        self._resources = []
        for i in range(20):
            r = self._svc.register_resource(
                type_name="git-checkout",
                name=f"bench/repo-{i}",
                location=f"/tmp/bench/repo-{i}",
                description=f"Benchmark repo {i}",
            )
            self._resources.append(r)

    def time_show_type_builtin(self) -> None:
        """Benchmark looking up a built-in type by name."""
        self._svc.show_type("git-checkout")

    def time_show_type_second_builtin(self) -> None:
        """Benchmark looking up the second built-in type."""
        self._svc.show_type("fs-directory")

    def time_list_types(self) -> None:
        """Benchmark listing all registered types."""
        self._svc.list_types()

    def time_list_types_filtered(self) -> None:
        """Benchmark listing types filtered by namespace."""
        self._svc.list_types(namespace="builtin")

    def time_show_resource_by_name(self) -> None:
        """Benchmark looking up a resource by namespaced name."""
        self._svc.show_resource("bench/repo-0")

    def time_show_resource_by_ulid(self) -> None:
        """Benchmark looking up a resource by ULID."""
        self._svc.show_resource(self._resources[0].resource_id)

    def time_list_resources_all(self) -> None:
        """Benchmark listing all resources."""
        self._svc.list_resources()

    def time_list_resources_filtered(self) -> None:
        """Benchmark listing resources filtered by type."""
        self._svc.list_resources(type_name="git-checkout")


class RegistryRegistrationSuite:
    """Benchmark resource registration throughput."""

    _reg_ctr: int = 0

    def setup(self) -> None:
        self._sf = _setup_db()
        self._svc = ResourceRegistryService(session_factory=self._sf)
        self._svc.bootstrap_builtin_types()

    def time_register_resource(self) -> None:
        """Benchmark registering a single resource."""
        RegistryRegistrationSuite._reg_ctr += 1
        self._svc.register_resource(
            type_name="git-checkout",
            name=f"bench/reg-{RegistryRegistrationSuite._reg_ctr}",
            location=f"/tmp/bench/reg-{RegistryRegistrationSuite._reg_ctr}",
        )

    def time_bootstrap_builtin_types_idempotent(self) -> None:
        """Benchmark idempotent bootstrap (types already exist)."""
        self._svc.bootstrap_builtin_types()