"""ASV benchmarks for plan table ORM round-trip performance. Measures the performance of: - LifecyclePlanModel.from_domain() construction from Plan domain objects - LifecyclePlanModel.to_domain() conversion back to domain objects - Bulk plan model construction for migration-volume workloads """ from __future__ import annotations import sys from pathlib import Path try: from cleveragents.domain.models.core.plan import ( AutomationProfileProvenance, AutomationProfileRef, InvariantSource, NamespacedName, Plan, PlanIdentity, PlanInvariant, PlanPhase, ProcessingState, ProjectLink, ) from cleveragents.infrastructure.database.models import LifecyclePlanModel except ModuleNotFoundError: sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.domain.models.core.plan import ( AutomationProfileProvenance, AutomationProfileRef, InvariantSource, NamespacedName, Plan, PlanIdentity, PlanInvariant, PlanPhase, ProcessingState, ProjectLink, ) from cleveragents.infrastructure.database.models import LifecyclePlanModel def _make_plan(suffix: str = "bench") -> Plan: """Create a fully-populated Plan for benchmarking.""" return Plan( identity=PlanIdentity( plan_id=f"01ARZ3NDEKTSV4RRFFQ69G5F{suffix[:2].upper().ljust(2, 'A')}" ), namespaced_name=NamespacedName( server=None, namespace="local", name=f"plan-{suffix}" ), description="Benchmark plan for ORM round-trip testing", action_name="local/bench-action", definition_of_done="All benchmarks pass within time budget", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, automation_profile=AutomationProfileRef( profile_name="org/bench-profile", provenance=AutomationProfileProvenance.ACTION, ), project_links=[ ProjectLink(project_name="local/project-a", alias="main"), ProjectLink(project_name="local/project-b", alias="lib", read_only=True), ], invariants=[ PlanInvariant( text="Global invariant", source=InvariantSource.GLOBAL, ), PlanInvariant( text="Action invariant", source=InvariantSource.ACTION, ), PlanInvariant( text="Plan invariant", source=InvariantSource.PLAN, ), ], strategy_actor="openai/gpt-4", execution_actor="openai/gpt-4", estimation_actor="openai/gpt-4", invariant_actor="openai/gpt-4", arguments={"target_coverage": 90, "framework": "pytest"}, arguments_order=["target_coverage", "framework"], created_by="bench-user", tags=["benchmark", "test"], reusable=True, read_only=False, ) class PlanModelFromDomainSuite: """Benchmark LifecyclePlanModel.from_domain() construction.""" def setup(self) -> None: self.plan = _make_plan() self.minimal_plan = Plan( identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"), namespaced_name=NamespacedName( server=None, namespace="local", name="min-plan" ), description="Minimal plan", action_name="local/min-action", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, ) def time_from_domain_full(self) -> None: """Convert a fully-populated Plan to ORM model.""" LifecyclePlanModel.from_domain(self.plan) def time_from_domain_minimal(self) -> None: """Convert a minimal Plan to ORM model.""" LifecyclePlanModel.from_domain(self.minimal_plan) class PlanModelToDomainSuite: """Benchmark LifecyclePlanModel.to_domain() conversion.""" def setup(self) -> None: self.full_model = LifecyclePlanModel.from_domain(_make_plan()) self.minimal_model = LifecyclePlanModel.from_domain( Plan( identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"), namespaced_name=NamespacedName( server=None, namespace="local", name="minimal" ), description="Minimal plan", action_name="local/min-action", phase=PlanPhase.STRATEGIZE, processing_state=ProcessingState.QUEUED, ) ) def time_to_domain_full(self) -> None: """Convert a fully-populated ORM model back to domain.""" self.full_model.to_domain() def time_to_domain_minimal(self) -> None: """Convert a minimal ORM model back to domain.""" self.minimal_model.to_domain() class PlanModelBulkSuite: """Benchmark bulk plan model construction for migration volumes.""" def setup(self) -> None: self.plans = [_make_plan(str(i)) for i in range(50)] def time_bulk_from_domain_50(self) -> None: """Convert 50 Plan domain objects to ORM models.""" for plan in self.plans: LifecyclePlanModel.from_domain(plan) def time_bulk_round_trip_50(self) -> None: """Full round-trip: domain -> ORM -> domain for 50 plans.""" for plan in self.plans: model = LifecyclePlanModel.from_domain(plan) model.to_domain()