"""ASV benchmarks for CRP domain model validation throughput. Measures the performance of: - ContextRequest creation and validation - ContextFragment creation and validation - ContextBudget creation and validation - DetailLevelMap resolution (integer and named) - AssembledContext creation """ 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 the top-level package so Python picks up the source tree # version instead of the potentially stale installed copy. import cleveragents # noqa: E402 importlib.reload(cleveragents) from cleveragents.domain.models.acms.crp import ( # noqa: E402 AssembledContext, ContextBudget, ContextFragment, ContextRequest, DetailLevelMap, FragmentProvenance, ) class TimeCRPModelCreation: """Benchmark CRP model creation throughput.""" timeout = 60 def setup(self) -> None: """Prepare reusable inputs.""" self.provenance = FragmentProvenance( resource_uri="uko-py:module/auth", location="lines 1-50", strategy="simple-keyword", ) self.dlm = DetailLevelMap( domain="uko-code:", max_depth=9, levels={ "MODULE_LISTING": 0, "MODULE_GRAPH": 1, "MEMBER_LISTING": 2, "MEMBER_SUMMARY": 3, "SIGNATURES": 4, "SIGNATURES_WITH_DOCS": 5, "STRUCTURAL_OUTLINE": 6, "KEY_LOGIC": 7, "NEAR_COMPLETE": 8, "FULL_SOURCE": 9, }, ) def time_context_request_defaults(self) -> None: """Create ContextRequest with defaults.""" for _ in range(1000): ContextRequest(purpose="benchmark") def time_context_request_full(self) -> None: """Create ContextRequest with all fields populated.""" for _ in range(1000): ContextRequest( query="auth flow", entities=["AuthManager"], uko_types=["uko-py:Class"], focus=["uko-py:class/AuthManager"], breadth=3, depth=4, depth_gradient=True, max_tokens=8000, preferred_strategies=["arce", "semantic-embedding"], required_backends=["vector", "graph"], priority=0.8, purpose="Understand auth flow", ) def time_context_fragment(self) -> None: """Create ContextFragment with provenance.""" for _ in range(1000): ContextFragment( uko_node="uko-py:class/AuthManager", content="class AuthManager: ...", detail_depth=9, token_count=800, relevance_score=0.95, provenance=self.provenance, ) def time_context_budget(self) -> None: """Create ContextBudget and access available_tokens.""" for _ in range(1000): b = ContextBudget(max_tokens=8000, reserved_tokens=1000) _ = b.available_tokens def time_detail_level_map_resolve_int(self) -> None: """Resolve integer depths via DetailLevelMap.""" for i in range(1000): self.dlm.resolve(i % 15) def time_detail_level_map_resolve_named(self) -> None: """Resolve named levels via DetailLevelMap.""" names: list[str] = list(self.dlm.levels.keys()) for i in range(1000): self.dlm.resolve(names[i % len(names)]) def time_assembled_context(self) -> None: """Create AssembledContext.""" for _ in range(1000): AssembledContext( total_tokens=1500, budget_used=0.75, context_hash="abc123", strategies_used=["simple-keyword", "arce"], ) def time_fragment_provenance(self) -> None: """Create FragmentProvenance.""" for _ in range(1000): FragmentProvenance( resource_uri="uko-py:module/auth", location="lines 1-50", strategy="arce", )