cleveragents-core/benchmarks/uko_indexer_bench.py

"""ASV benchmarks for UKO Indexer Real-time Index Synchronization.

Measures the performance of:
- UKOIndexer.index_resource() pipeline (analyze → provenance → graph → text → vector)
- UKOIndexer.remove_resource() cleanup
- UKOIndexer.reindex_resource() lifecycle
- In-memory index backend operations
- Graceful degradation (no optional backends)
"""

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)

# Force-reload so ASV picks up the source tree version.
import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.application.services.uko_indexer import (  # noqa: E402
    UKOIndexer,
)
from cleveragents.domain.models.acms.analyzers import (  # noqa: E402
    AnalyzerRegistry,
)
from cleveragents.domain.models.acms.index_stubs import (  # noqa: E402
    InMemoryGraphIndexBackend,
    InMemoryTextIndexBackend,
    InMemoryVectorIndexBackend,
)
from cleveragents.domain.models.acms.python_analyzer import (  # noqa: E402
    PythonAnalyzer,
)
from cleveragents.domain.models.core.resource import Resource  # noqa: E402

# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------

ULID_1 = "01HQ8ZDRX50000000000000001"
ULID_2 = "01HQ8ZDRX50000000000000002"
PROJECT = "local/bench"

_SMALL_PYTHON = 'import os\n\ndef hello():\n    """Say hello."""\n    pass\n'

_MEDIUM_PYTHON = (
    '"""Module docstring."""\n\nimport os\nfrom pathlib import Path\n\n'
    + "\n\n".join(
        f'class Cls{i}:\n    """Class {i}."""\n'
        f"    def method_{i}(self):\n"
        f'        """Method {i}."""\n        pass\n'
        for i in range(20)
    )
)


class _InMemoryContentReader:
    """Content reader for benchmarks."""

    def __init__(self) -> None:
        self._content: dict[str, str] = {}

    def set_content(self, resource_id: str, content: str) -> None:
        self._content[resource_id] = content

    def read_content(self, resource: Resource) -> str:
        return self._content[resource.resource_id]


def _make_resource(resource_id: str) -> Resource:
    return Resource(
        resource_id=resource_id,
        resource_type_name="git-checkout",
        classification="physical",
        location="src/bench.py",
    )


# ---------------------------------------------------------------------------
# Index pipeline benchmarks
# ---------------------------------------------------------------------------


class IndexResourceSuite:
    """Benchmark UKOIndexer.index_resource() pipeline."""

    timeout = 120

    def setup(self) -> None:
        self.resource = _make_resource(ULID_1)

    def time_index_small_python(self) -> None:
        # Re-create indexer each time to avoid accumulation
        registry = AnalyzerRegistry()
        registry.register(PythonAnalyzer())
        reader = _InMemoryContentReader()
        reader.set_content(ULID_1, _SMALL_PYTHON)
        indexer = UKOIndexer(
            analyzer_registry=registry,
            graph_backend=InMemoryGraphIndexBackend(),
            text_backend=InMemoryTextIndexBackend(),
            vector_backend=InMemoryVectorIndexBackend(),
            content_reader=reader,
        )
        indexer.index_resource(self.resource, project=PROJECT)

    def time_index_medium_python(self) -> None:
        registry = AnalyzerRegistry()
        registry.register(PythonAnalyzer())
        reader = _InMemoryContentReader()
        reader.set_content(ULID_1, _MEDIUM_PYTHON)
        indexer = UKOIndexer(
            analyzer_registry=registry,
            graph_backend=InMemoryGraphIndexBackend(),
            text_backend=InMemoryTextIndexBackend(),
            vector_backend=InMemoryVectorIndexBackend(),
            content_reader=reader,
        )
        indexer.index_resource(self.resource, project=PROJECT)


class IndexGracefulDegradationSuite:
    """Benchmark indexing without optional backends."""

    timeout = 120

    def setup(self) -> None:
        self.resource = _make_resource(ULID_1)

    def time_index_graph_only(self) -> None:
        registry = AnalyzerRegistry()
        registry.register(PythonAnalyzer())
        reader = _InMemoryContentReader()
        reader.set_content(ULID_1, _SMALL_PYTHON)
        indexer = UKOIndexer(
            analyzer_registry=registry,
            graph_backend=InMemoryGraphIndexBackend(),
            content_reader=reader,
        )
        indexer.index_resource(self.resource, project=PROJECT)


# ---------------------------------------------------------------------------
# Lifecycle benchmarks
# ---------------------------------------------------------------------------


class LifecycleSuite:
    """Benchmark index lifecycle (add/remove/reindex)."""

    timeout = 120

    def setup(self) -> None:
        self.resource = _make_resource(ULID_1)

    def time_remove_resource(self) -> None:
        registry = AnalyzerRegistry()
        registry.register(PythonAnalyzer())
        reader = _InMemoryContentReader()
        reader.set_content(ULID_1, _SMALL_PYTHON)
        indexer = UKOIndexer(
            analyzer_registry=registry,
            graph_backend=InMemoryGraphIndexBackend(),
            text_backend=InMemoryTextIndexBackend(),
            vector_backend=InMemoryVectorIndexBackend(),
            content_reader=reader,
        )
        indexer.index_resource(self.resource, project=PROJECT)
        indexer.remove_resource(ULID_1, project=PROJECT)

    def time_reindex_resource(self) -> None:
        registry = AnalyzerRegistry()
        registry.register(PythonAnalyzer())
        reader = _InMemoryContentReader()
        reader.set_content(ULID_1, _SMALL_PYTHON)
        indexer = UKOIndexer(
            analyzer_registry=registry,
            graph_backend=InMemoryGraphIndexBackend(),
            text_backend=InMemoryTextIndexBackend(),
            vector_backend=InMemoryVectorIndexBackend(),
            content_reader=reader,
        )
        indexer.index_resource(self.resource, project=PROJECT)
        reader.set_content(ULID_1, _MEDIUM_PYTHON)
        indexer.reindex_resource(self.resource, project=PROJECT)


# ---------------------------------------------------------------------------
# Backend operation benchmarks
# ---------------------------------------------------------------------------


class BackendOperationsSuite:
    """Benchmark raw in-memory backend operations."""

    timeout = 60

    def time_text_index_and_search(self) -> None:
        be = InMemoryTextIndexBackend()
        be.index_document(PROJECT, "doc1", "hello world", {"k": "v"})
        be.search(PROJECT, "hello", limit=10)

    def time_vector_index_and_search(self) -> None:
        be = InMemoryVectorIndexBackend()
        be.index_embedding(PROJECT, "emb1", [1.0, 2.0, 3.0], {"k": "v"})
        be.search_similar(PROJECT, [1.0, 2.0, 3.0], limit=10)

    def time_graph_add_and_query(self) -> None:
        be = InMemoryGraphIndexBackend()
        be.add_triple(PROJECT, "s1", "p1", "o1")
        be.query(PROJECT, "SELECT *")

    def time_graph_add_and_remove(self) -> None:
        be = InMemoryGraphIndexBackend()
        be.add_triple(PROJECT, "s1", "p1", "o1")
        be.remove_triples(PROJECT, subject="s1", predicate=None, obj=None)