cleveragents-core/benchmarks/unified_context_models_bench.py

"""ASV benchmarks for unified CRP/core context model hierarchy.

Measures the construction throughput for core types that extend CRP base
types via Pydantic v2 inheritance, and the isinstance-check overhead.
"""

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 as CRPAssembledContext,
)
from cleveragents.domain.models.acms.crp import (  # noqa: E402
    ContextBudget as CRPContextBudget,
)
from cleveragents.domain.models.acms.crp import (  # noqa: E402
    ContextFragment as CRPContextFragment,
)
from cleveragents.domain.models.acms.crp import (  # noqa: E402
    FragmentProvenance as CRPFragmentProvenance,
)
from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
    ContextBudget,
    ContextFragment,
    ContextPayload,
    FragmentProvenance,
)


class TimeUnifiedModelCreation:
    """Benchmark unified model construction throughput."""

    timeout = 60

    def setup(self) -> None:
        """Prepare reusable inputs."""
        self.provenance = FragmentProvenance(
            resource_uri="uko-py:module/bench",
            strategy="benchmark-strategy",
            resource_type="git-checkout",
        )

    def time_fragment_provenance(self) -> None:
        """Create core FragmentProvenance (inherits CRP)."""
        for _ in range(1000):
            FragmentProvenance(
                resource_uri="uko-py:module/bench",
                strategy="arce",
                resource_type="git-checkout",
            )

    def time_context_fragment(self) -> None:
        """Create core ContextFragment (inherits CRP)."""
        for _ in range(1000):
            ContextFragment(
                uko_node="project://bench",
                content="benchmark fragment",
                token_count=100,
                tier="hot",
                provenance=self.provenance,
            )

    def time_context_budget(self) -> None:
        """Create core ContextBudget (inherits CRP)."""
        for _ in range(1000):
            b = ContextBudget(max_tokens=8192, reserved_tokens=1024)
            _ = b.available_tokens

    def time_context_payload(self) -> None:
        """Create core ContextPayload (inherits CRP AssembledContext)."""
        for _ in range(1000):
            ContextPayload(plan_id="01JQBENCHPN0000000000000AA")


class TimeIsinstanceChecks:
    """Benchmark isinstance compatibility checks."""

    timeout = 60

    def setup(self) -> None:
        """Prepare instances."""
        prov = FragmentProvenance(resource_uri="uko-py:module/bench")
        self.provenance = prov
        self.fragment = ContextFragment(
            uko_node="project://bench",
            content="benchmark fragment",
            token_count=100,
            provenance=prov,
        )
        self.budget = ContextBudget(max_tokens=4096, reserved_tokens=512)
        self.payload = ContextPayload(
            plan_id="01JQBENCHPN0000000000000AA",
        )

    def time_isinstance_provenance(self) -> None:
        """isinstance check: FragmentProvenance against CRP base."""
        for _ in range(10000):
            isinstance(self.provenance, CRPFragmentProvenance)

    def time_isinstance_fragment(self) -> None:
        """isinstance check: ContextFragment against CRP base."""
        for _ in range(10000):
            isinstance(self.fragment, CRPContextFragment)

    def time_isinstance_budget(self) -> None:
        """isinstance check: ContextBudget against CRP base."""
        for _ in range(10000):
            isinstance(self.budget, CRPContextBudget)

    def time_isinstance_payload(self) -> None:
        """isinstance check: ContextPayload against CRP AssembledContext."""
        for _ in range(10000):
            isinstance(self.payload, CRPAssembledContext)