cleveragents-core/benchmarks/scoped_view_bench.py

"""ASV benchmarks for ACMS Scoped Backend View filtering.

Measures the performance of:
- ResourceScope creation and containment checks
- ResourceScope path matching with include/exclude globs
- ScopedBackendView fragment visibility filtering
- ScopedBackendView backend search proxying overhead
- ScopedBackendSet creation and multi-backend search
- ResourceAliasResolver alias resolution
- resolve_resource_scope allowlist/denylist resolution
- validate_project_scope and validate_resource_scope guards
"""

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.domain.models.acms.scope_resolution import (  # noqa: E402
    resolve_resource_scope,
    validate_project_scope,
    validate_resource_scope,
)
from cleveragents.domain.models.acms.scoped_view import (  # noqa: E402
    ResourceScope,
    ScopedBackendView,
    create_scoped_backend_set,
)
from cleveragents.domain.models.acms.stubs import (  # noqa: E402
    InMemoryGraphBackend,
    InMemoryTextBackend,
    InMemoryVectorBackend,
)
from cleveragents.domain.models.acms.tiers import (  # noqa: E402
    ContextTier,
    TieredFragment,
)
from cleveragents.domain.models.core.project import (  # noqa: E402
    LinkedResource,
    NamespacedProject,
)

# ---------------------------------------------------------------------------
# ResourceScope benchmarks
# ---------------------------------------------------------------------------


class ResourceScopeCreationSuite:
    """Benchmark ResourceScope instantiation overhead."""

    timeout = 60

    def time_create_small_scope(self) -> None:
        ResourceScope(
            resource_ids=frozenset({"R1", "R2"}),
            project_names=frozenset({"local/api"}),
        )

    def time_create_large_scope(self) -> None:
        ResourceScope(
            resource_ids=frozenset(f"R{i:04d}" for i in range(1000)),
            project_names=frozenset(f"ns/proj{i}" for i in range(10)),
        )

    def time_contains_resource(self) -> None:
        scope = ResourceScope(
            resource_ids=frozenset(f"R{i:04d}" for i in range(1000)),
            project_names=frozenset({"p1"}),
        )
        scope.contains_resource("R0500")

    def time_contains_project(self) -> None:
        scope = ResourceScope(
            resource_ids=frozenset({"R1"}),
            project_names=frozenset(f"ns/proj{i}" for i in range(100)),
        )
        scope.contains_project("ns/proj50")


# ---------------------------------------------------------------------------
# Path matching benchmarks
# ---------------------------------------------------------------------------


class PathMatchingSuite:
    """Benchmark ResourceScope.matches_path() with various patterns."""

    timeout = 60

    def setup(self) -> None:
        self.scope_include = ResourceScope(
            resource_ids=frozenset({"R1"}),
            project_names=frozenset({"p1"}),
            include_paths=("src/**/*.py", "tests/**/*.py"),
        )
        self.scope_exclude = ResourceScope(
            resource_ids=frozenset({"R1"}),
            project_names=frozenset({"p1"}),
            exclude_paths=("**/node_modules/**", "**/__pycache__/**"),
        )
        self.scope_both = ResourceScope(
            resource_ids=frozenset({"R1"}),
            project_names=frozenset({"p1"}),
            include_paths=("src/**/*.py",),
            exclude_paths=("**/test_*",),
        )

    def time_match_include_hit(self) -> None:
        self.scope_include.matches_path("src/models/user.py")

    def time_match_include_miss(self) -> None:
        self.scope_include.matches_path("docs/readme.md")

    def time_match_exclude_pass(self) -> None:
        self.scope_exclude.matches_path("src/main.py")

    def time_match_exclude_reject(self) -> None:
        self.scope_exclude.matches_path("node_modules/foo/bar.js")

    def time_match_combined(self) -> None:
        self.scope_both.matches_path("src/main.py")


# ---------------------------------------------------------------------------
# ScopedBackendView benchmarks
# ---------------------------------------------------------------------------


class FragmentFilteringSuite:
    """Benchmark ScopedBackendView.is_visible() fragment filtering."""

    timeout = 60

    def setup(self) -> None:
        self.view = ScopedBackendView(
            allowed_projects=frozenset({"proj-a"}),
            resource_ids=frozenset({"R1", "R2"}),
        )
        self.frag_visible = TieredFragment(
            fragment_id="f1",
            content="c",
            tier=ContextTier("hot"),
            project_name="proj-a",
            resource_id="R1",
        )
        self.frag_hidden = TieredFragment(
            fragment_id="f2",
            content="c",
            tier=ContextTier("hot"),
            project_name="proj-b",
            resource_id="R1",
        )

    def time_visible_fragment(self) -> None:
        self.view.is_visible(self.frag_visible)

    def time_hidden_fragment(self) -> None:
        self.view.is_visible(self.frag_hidden)

    def time_effective_scope(self) -> None:
        self.view.effective_scope()


class BackendProxyingSuite:
    """Benchmark ScopedBackendView search proxying overhead."""

    timeout = 60

    def setup(self) -> None:
        scope = ResourceScope(
            resource_ids=frozenset({"R1", "R2"}),
            project_names=frozenset({"local/api"}),
        )
        self.view = ScopedBackendView.from_resource_scope(scope)
        self.text_backend = InMemoryTextBackend()
        self.vector_backend = InMemoryVectorBackend()
        self.graph_backend = InMemoryGraphBackend()
        self.embedding: list[float] = [0.1] * 768

    def time_search_text(self) -> None:
        self.view.search_text(self.text_backend, "auth flow")

    def time_search_vector(self) -> None:
        self.view.search_vector(self.vector_backend, self.embedding)

    def time_search_graph(self) -> None:
        self.view.search_graph(
            self.graph_backend, "SELECT ?s WHERE { ?s a uko:Container }"
        )


# ---------------------------------------------------------------------------
# ScopedBackendSet benchmarks
# ---------------------------------------------------------------------------


class BackendSetSuite:
    """Benchmark ScopedBackendSet creation and multi-backend search."""

    timeout = 60

    def setup(self) -> None:
        self.scope = ResourceScope(
            resource_ids=frozenset({"R1"}),
            project_names=frozenset({"local/api"}),
        )

    def time_create_backend_set(self) -> None:
        create_scoped_backend_set(
            self.scope,
            text_backend=InMemoryTextBackend(),
            vector_backend=InMemoryVectorBackend(),
            graph_backend=InMemoryGraphBackend(),
        )

    def time_backend_set_text_search(self) -> None:
        bset = create_scoped_backend_set(self.scope, text_backend=InMemoryTextBackend())
        bset.search_text("query")


# ---------------------------------------------------------------------------
# Scope resolution benchmarks
# ---------------------------------------------------------------------------


class ScopeResolutionSuite:
    """Benchmark resolve_resource_scope and validation guards."""

    timeout = 60

    def setup(self) -> None:
        self.project = NamespacedProject(
            name="api",
            namespace="local",
            linked_resources=[
                LinkedResource(resource_id="01HQ8ZDRX50000000000000001", alias="repo"),
                LinkedResource(resource_id="01HQ8ZDRX50000000000000002", alias="docs"),
            ],
        )
        self.scope = ResourceScope(
            resource_ids=frozenset({"R1", "R2", "R3"}),
            project_names=frozenset({"p1", "p2"}),
        )

    def time_resolve_scope_no_filter(self) -> None:
        resolve_resource_scope([self.project])

    def time_resolve_scope_with_include(self) -> None:
        resolve_resource_scope([self.project], include_resources=("repo",))

    def time_resolve_scope_with_exclude(self) -> None:
        resolve_resource_scope([self.project], exclude_resources=("docs",))

    def time_validate_project_scope(self) -> None:
        validate_project_scope(frozenset({"p1"}), frozenset({"p1", "p2"}))

    def time_validate_resource_scope(self) -> None:
        validate_resource_scope(frozenset({"R1"}), self.scope)