cleveragents-core/benchmarks/large_project_decompose_bench.py

"""ASV benchmarks for large-project decomposition overhead.

Measures the time to decompose file sets of various sizes and to
compute dependency closures on synthetic graphs.
"""

from __future__ import annotations

import importlib
import os
import sys
import tempfile
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)

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

import itertools  # noqa: E402

from cleveragents.application.services.decomposition_graph import (  # noqa: E402
    DependencyClosureComputer,
)
from cleveragents.application.services.decomposition_models import (  # noqa: E402
    DecompositionConfig,
    DependencyEdge,
    DependencyGraph,
    DependencyType,
)
from cleveragents.application.services.decomposition_service import (  # noqa: E402
    DecompositionService,
)


def _build_files(tmpdir: str, count: int, levels: int = 3) -> list[str]:
    """Generate *count* files in *tmpdir* across *levels* dirs."""
    paths: list[str] = []
    for i in range(count):
        level = i % levels
        parts = [f"level{d}" for d in range(level + 1)]
        dirpath = os.path.join(tmpdir, *parts)
        os.makedirs(dirpath, exist_ok=True)
        fpath = os.path.join(dirpath, f"file_{i:06d}.py")
        with open(fpath, "w") as fh:
            fh.write("x" * 200)
        paths.append(fpath)
    return sorted(paths)


def _build_chain_graph(size: int) -> DependencyGraph:
    """Build a chain of *size* nodes."""
    g = DependencyGraph()
    nodes = [f"file_{i:06d}.py" for i in range(size)]
    for n in nodes:
        g.add_node(n)
    for src, tgt in itertools.pairwise(nodes):
        g.add_edge(DependencyEdge(src, tgt, DependencyType.IMPORT))
    return g, nodes  # type: ignore[return-value]


class DecomposeSmallSuite:
    """Benchmark decomposition of a small file set (100 files)."""

    def setup(self) -> None:
        self._tmpdir = tempfile.mkdtemp(prefix="bench-decompose-")
        self._files = _build_files(self._tmpdir, 100)
        self._svc = DecompositionService()

    def time_decompose_default(self) -> None:
        """Decompose 100 files with default config."""
        self._svc.decompose(self._files)


class DecomposeMediumSuite:
    """Benchmark decomposition of a medium file set (1000 files)."""

    def setup(self) -> None:
        self._tmpdir = tempfile.mkdtemp(prefix="bench-decompose-")
        self._files = _build_files(self._tmpdir, 1000)
        self._svc = DecompositionService()

    def time_decompose_default(self) -> None:
        """Decompose 1000 files with default config."""
        self._svc.decompose(self._files)


class DecomposeLargeSuite:
    """Benchmark decomposition of a large file set (5000 files)."""

    def setup(self) -> None:
        self._tmpdir = tempfile.mkdtemp(prefix="bench-decompose-")
        self._files = _build_files(self._tmpdir, 5000, levels=5)
        self._svc = DecompositionService()
        self._config = DecompositionConfig(max_files_per_subplan=200)

    def time_decompose_custom(self) -> None:
        """Decompose 5000 files with custom config."""
        self._svc.decompose(self._files, self._config)


class ClosureSuite:
    """Benchmark dependency closure computation."""

    def setup(self) -> None:
        result = _build_chain_graph(1000)
        self._graph: DependencyGraph = result[0]  # type: ignore[assignment]
        self._roots: list[str] = [result[1][0]]  # type: ignore[index]
        self._computer = DependencyClosureComputer()

    def time_closure_unbounded(self) -> None:
        """Compute unbounded closure on a 1000-node chain."""
        self._computer.clear_cache()
        self._computer.compute_closure(self._graph, self._roots)

    def time_closure_bounded(self) -> None:
        """Compute bounded closure (cutoff=100) on a 1000-node chain."""
        self._computer.clear_cache()
        self._computer.compute_closure(self._graph, self._roots, cutoff=100)

    def time_closure_memoized(self) -> None:
        """Compute closure twice — second should be memoized."""
        self._computer.clear_cache()
        self._computer.compute_closure(self._graph, self._roots)
        self._computer.compute_closure(self._graph, self._roots)