cleveragents-core/benchmarks/invariant_reconciliation_bench.py

"""ASV benchmarks for Invariant Reconciliation Actor throughput.

Measures the performance of:
- Reconciliation with varying numbers of invariants per scope
- Conflict detection and resolution overhead
- Decision recording throughput
- Full pipeline (collect → reconcile → record)
"""

from __future__ import annotations

import importlib
import sys
from pathlib import Path
from typing import ClassVar

# Ensure the local source tree is importable
_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)

from cleveragents.actor.reconciliation import (  # noqa: E402
    InvariantReconciliationActor,
    ScopeInvariants,
    reconcile_invariants,
)
from cleveragents.application.services.decision_service import (  # noqa: E402
    DecisionService,
)
from cleveragents.application.services.invariant_service import (  # noqa: E402
    InvariantService,
)
from cleveragents.domain.models.core.invariant import (  # noqa: E402
    Invariant,
    InvariantScope,
)


def _make_invariant(text: str, scope: InvariantScope, source: str) -> Invariant:
    """Create an invariant without going through the service."""
    return Invariant(text=text, scope=scope, source_name=source)


def _build_scope_invariants(
    n_global: int = 0,
    n_project: int = 0,
    n_action: int = 0,
    n_plan: int = 0,
) -> ScopeInvariants:
    """Build a ScopeInvariants with the given counts (unique texts)."""
    return ScopeInvariants(
        global_invariants=[
            _make_invariant(f"global-rule-{i}", InvariantScope.GLOBAL, "system")
            for i in range(n_global)
        ],
        project_invariants=[
            _make_invariant(f"project-rule-{i}", InvariantScope.PROJECT, "local/api")
            for i in range(n_project)
        ],
        action_invariants=[
            _make_invariant(f"action-rule-{i}", InvariantScope.ACTION, "local/deploy")
            for i in range(n_action)
        ],
        plan_invariants=[
            _make_invariant(f"plan-rule-{i}", InvariantScope.PLAN, "PLAN01")
            for i in range(n_plan)
        ],
    )


def _build_conflicting_scope_invariants(n_conflicts: int) -> ScopeInvariants:
    """Build scope invariants where each text appears at global and plan."""
    return ScopeInvariants(
        global_invariants=[
            _make_invariant(f"shared-rule-{i}", InvariantScope.GLOBAL, "system")
            for i in range(n_conflicts)
        ],
        project_invariants=[],
        action_invariants=[],
        plan_invariants=[
            _make_invariant(f"shared-rule-{i}", InvariantScope.PLAN, "PLAN01")
            for i in range(n_conflicts)
        ],
    )


class ReconciliationAlgorithmSuite:
    """Benchmark the pure reconciliation algorithm (no I/O)."""

    params: ClassVar[list[int]] = [1, 10, 50, 100]
    param_names: ClassVar[list[str]] = ["invariants_per_scope"]

    def setup(self, n: int) -> None:
        self._scope_invs = _build_scope_invariants(
            n_global=n, n_project=n, n_action=n, n_plan=n
        )

    def time_reconcile_no_conflicts(self, n: int) -> None:
        """Reconcile N invariants per scope with no conflicts."""
        reconcile_invariants(self._scope_invs)


class ConflictResolutionSuite:
    """Benchmark conflict detection and resolution."""

    params: ClassVar[list[int]] = [1, 10, 50, 100]
    param_names: ClassVar[list[str]] = ["conflict_count"]

    def setup(self, n: int) -> None:
        self._scope_invs = _build_conflicting_scope_invariants(n)

    def time_reconcile_with_conflicts(self, n: int) -> None:
        """Reconcile N conflicting invariant pairs."""
        reconcile_invariants(self._scope_invs)


class FullPipelineSuite:
    """Benchmark the full actor pipeline (collect + reconcile + record)."""

    params: ClassVar[list[int]] = [1, 10, 50]
    param_names: ClassVar[list[str]] = ["invariants_per_scope"]

    def setup(self, n: int) -> None:
        self._n = n

    def time_full_pipeline(self, n: int) -> None:
        """Run full actor pipeline with N invariants per scope."""
        inv_svc = InvariantService()
        dec_svc = DecisionService()

        for i in range(n):
            inv_svc.add_invariant(f"global-rule-{i}", InvariantScope.GLOBAL, "system")
            inv_svc.add_invariant(
                f"project-rule-{i}", InvariantScope.PROJECT, "local/api"
            )
            inv_svc.add_invariant(
                f"action-rule-{i}", InvariantScope.ACTION, "local/deploy"
            )
            inv_svc.add_invariant(
                f"plan-rule-{i}", InvariantScope.PLAN, "01JQAAAAAAAAAAAAAAAAAAAA01"
            )

        actor = InvariantReconciliationActor(
            invariant_service=inv_svc,
            decision_service=dec_svc,
        )
        actor.run(
            plan_id="01JQAAAAAAAAAAAAAAAAAAAA01",
            project_name="local/api",
            action_name="local/deploy",
        )


class DecisionRecordingSuite:
    """Benchmark decision recording throughput."""

    params: ClassVar[list[int]] = [1, 10, 50, 100]
    param_names: ClassVar[list[str]] = ["invariant_count"]

    def setup(self, n: int) -> None:
        self._n = n

    def time_record_decisions(self, n: int) -> None:
        """Record N invariant_enforced decisions."""
        inv_svc = InvariantService()
        dec_svc = DecisionService()

        for i in range(n):
            inv_svc.add_invariant(f"global-rule-{i}", InvariantScope.GLOBAL, "system")

        actor = InvariantReconciliationActor(
            invariant_service=inv_svc,
            decision_service=dec_svc,
        )
        actor.run(plan_id="01JQAAAAAAAAAAAAAAAAAAAA01")