cleveragents-core/benchmarks/sandbox_boundary_bench.py

"""ASV benchmarks for sandbox boundary algebra.

Measures the performance of:
- ``is_sandbox_boundary()`` predicate
- ``sandbox_boundary()`` walk for shallow and deep DAGs
- ``compute_sandbox_domains()`` grouping
- ``BoundaryCache`` lookup hit and miss
"""

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.core.resource import (  # noqa: E402
    PhysVirt,
    Resource,
    ResourceCapabilities,
    SandboxStrategy,
)
from cleveragents.infrastructure.sandbox.boundary import (  # noqa: E402
    BoundaryCache,
    compute_sandbox_domains,
    is_sandbox_boundary,
    sandbox_boundary,
)

# ---------------------------------------------------------------------------
# Test data builders
# ---------------------------------------------------------------------------

_COUNTER = 0


def _ulid(name: str) -> str:
    global _COUNTER
    _COUNTER += 1
    base = abs(hash(name)) % (10**20)
    raw = f"{_COUNTER:06d}{base:020d}"
    charset = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
    result = ""
    for ch in raw:
        result += charset[int(ch)]
    return (result + "0" * 26)[:26]


def _resource(
    name: str,
    sandboxable: bool = False,
    strategy: str = "none",
    parents: list[str] | None = None,
) -> Resource:
    return Resource(
        resource_id=_ulid(name),
        name=name,
        resource_type_name="git-checkout" if sandboxable else "fs-file",
        classification=PhysVirt.PHYSICAL,
        sandbox_strategy=SandboxStrategy(strategy),
        capabilities=ResourceCapabilities(
            readable=True,
            writable=True,
            sandboxable=sandboxable,
            checkpointable=False,
        ),
        parents=parents or [],
    )


# ---------------------------------------------------------------------------
# Pre-built test fixtures
# ---------------------------------------------------------------------------


def _build_shallow_dag() -> tuple[Resource, dict[str, Resource]]:
    """Build a shallow DAG: boundary -> 10 files."""
    boundary = _resource("bench-co", sandboxable=True, strategy="git_worktree")
    registry = {boundary.resource_id: boundary}
    leaf = boundary
    for i in range(10):
        child = _resource(f"bench-f{i}", parents=[boundary.resource_id])
        registry[child.resource_id] = child
        leaf = child
    return leaf, registry


def _build_deep_dag(depth: int = 50) -> tuple[Resource, dict[str, Resource]]:
    """Build a deep linear DAG: boundary -> dir1 -> dir2 -> ... -> file."""
    boundary = _resource("bench-deep-co", sandboxable=True, strategy="git_worktree")
    registry = {boundary.resource_id: boundary}
    current = boundary
    for i in range(depth):
        child = _resource(f"bench-d{i}", parents=[current.resource_id])
        registry[child.resource_id] = child
        current = child
    return current, registry


_SHALLOW_LEAF, _SHALLOW_REGISTRY = _build_shallow_dag()
_DEEP_LEAF, _DEEP_REGISTRY = _build_deep_dag()


# ---------------------------------------------------------------------------
# is_sandbox_boundary benchmarks
# ---------------------------------------------------------------------------


class TimeSandboxBoundaryCheck:
    """Benchmark is_sandbox_boundary() predicate."""

    def setup(self) -> None:
        self.boundary = _resource("bm-co", sandboxable=True, strategy="git_worktree")
        self.non_boundary = _resource("bm-f", sandboxable=False)

    def time_is_boundary_true(self) -> None:
        is_sandbox_boundary(self.boundary)

    def time_is_boundary_false(self) -> None:
        is_sandbox_boundary(self.non_boundary)


# ---------------------------------------------------------------------------
# sandbox_boundary() walk benchmarks
# ---------------------------------------------------------------------------


class TimeSandboxBoundaryWalk:
    """Benchmark sandbox_boundary() for various DAG shapes."""

    def time_self_boundary(self) -> None:
        """Walk when resource is itself the boundary (O(1))."""
        boundary = _SHALLOW_REGISTRY[_SHALLOW_LEAF.parents[0]]
        sandbox_boundary(boundary, _SHALLOW_REGISTRY)

    def time_shallow_walk(self) -> None:
        """Walk one hop from leaf to boundary."""
        sandbox_boundary(_SHALLOW_LEAF, _SHALLOW_REGISTRY)

    def time_deep_walk(self) -> None:
        """Walk 50 hops from leaf to boundary."""
        sandbox_boundary(_DEEP_LEAF, _DEEP_REGISTRY)


# ---------------------------------------------------------------------------
# compute_sandbox_domains benchmarks
# ---------------------------------------------------------------------------


class TimeSandboxDomains:
    """Benchmark compute_sandbox_domains() grouping."""

    def setup(self) -> None:
        self.boundary = _resource("dom-co", sandboxable=True, strategy="git_worktree")
        self.resources = []
        self.registry = {self.boundary.resource_id: self.boundary}
        for i in range(100):
            r = _resource(f"dom-f{i}", parents=[self.boundary.resource_id])
            self.resources.append(r)
            self.registry[r.resource_id] = r

    def time_domains_100_resources(self) -> None:
        compute_sandbox_domains(self.resources, self.registry)


# ---------------------------------------------------------------------------
# BoundaryCache benchmarks
# ---------------------------------------------------------------------------


class TimeBoundaryCache:
    """Benchmark BoundaryCache hit and miss paths."""

    def setup(self) -> None:
        self.cache = BoundaryCache()
        self.leaf = _SHALLOW_LEAF
        self.registry = _SHALLOW_REGISTRY

    def time_cache_miss(self) -> None:
        """First lookup (cache miss)."""
        self.cache.clear()
        self.cache.get_boundary(self.leaf, self.registry)

    def time_cache_hit(self) -> None:
        """Second lookup (cache hit)."""
        self.cache.get_boundary(self.leaf, self.registry)