"""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)