cleveragents-core/benchmarks/bench_fix_revalidate.py

"""ASV benchmarks for the Fix-then-Revalidate orchestration loop.

Measures the performance of:
- Orchestrator creation
- Fix loop with immediate pass
- Fix loop with multiple retries
- Fix loop with terminal failure
- Model creation and serialization
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

try:
    from cleveragents.application.services.fix_then_revalidate import (
        FixAttemptRecord,
        FixThenRevalidateOrchestrator,
        FixThenRevalidateResult,
    )
    from cleveragents.application.services.validation_pipeline import (
        ValidationPipeline,
        ValidationResult,
    )
    from cleveragents.domain.models.core.tool import ValidationMode
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.application.services.fix_then_revalidate import (
        FixAttemptRecord,
        FixThenRevalidateOrchestrator,
        FixThenRevalidateResult,
    )
    from cleveragents.application.services.validation_pipeline import (
        ValidationPipeline,
        ValidationResult,
    )
    from cleveragents.domain.models.core.tool import ValidationMode

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _mock_executor(validation_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    return {"passed": True, "message": f"{validation_name} ok"}


def _make_failed_result(name: str) -> ValidationResult:
    return ValidationResult(
        validation_name=name,
        resource_id="RES0001",
        resource_name="bench-resource",
        mode=ValidationMode.REQUIRED,
        passed=False,
        message=f"{name} failed",
        data={"hint": "bench"},
        duration_ms=1.0,
    )


def _always_fix(result: ValidationResult) -> str:
    return f"Fixed {result.validation_name}"


class _CountingRevalidate:
    def __init__(self, pass_after: int) -> None:
        self._pass_after = pass_after
        self._count = 0

    def __call__(self, result: ValidationResult) -> ValidationResult:
        self._count += 1
        passed = self._count >= self._pass_after
        return ValidationResult(
            validation_name=result.validation_name,
            resource_id=result.resource_id,
            resource_name=result.resource_name,
            mode=result.mode,
            passed=passed,
            message="passed" if passed else "failing",
            data=result.data,
            duration_ms=1.0,
        )


class _NeverPassRevalidate:
    def __call__(self, result: ValidationResult) -> ValidationResult:
        return ValidationResult(
            validation_name=result.validation_name,
            resource_id=result.resource_id,
            resource_name=result.resource_name,
            mode=result.mode,
            passed=False,
            message="failing",
            data=result.data,
            duration_ms=1.0,
        )


def _build_pipeline() -> ValidationPipeline:
    return ValidationPipeline(commands=[], executor=_mock_executor)


# ---------------------------------------------------------------------------
# Benchmark suites
# ---------------------------------------------------------------------------


class FixRevalidateCreationSuite:
    """Benchmarks for orchestrator instantiation."""

    timeout = 60

    def setup(self) -> None:
        self.pipeline = _build_pipeline()

    def time_create_orchestrator_default(self) -> None:
        """Benchmark creating an orchestrator with default settings."""
        FixThenRevalidateOrchestrator(
            validation_pipeline=self.pipeline,
        )

    def time_create_orchestrator_custom(self) -> None:
        """Benchmark creating an orchestrator with custom settings."""
        FixThenRevalidateOrchestrator(
            validation_pipeline=self.pipeline,
            max_retries=5,
            auto_strategy_revision=0.0,
        )


class FixRevalidateLoopSuite:
    """Benchmarks for the fix loop execution."""

    timeout = 60

    def setup(self) -> None:
        self.pipeline = _build_pipeline()

    def time_fix_first_attempt(self) -> None:
        """Benchmark fix loop that succeeds on first attempt."""
        orchestrator = FixThenRevalidateOrchestrator(
            validation_pipeline=self.pipeline,
            max_retries=3,
        )
        failed = [_make_failed_result("bench-check")]
        orchestrator.run_fix_loop(
            plan_id="BENCH001",
            failed_results=failed,
            fix_callback=_always_fix,
            revalidate_callback=_CountingRevalidate(pass_after=1),
        )

    def time_fix_on_second_attempt(self) -> None:
        """Benchmark fix loop that succeeds on the second attempt."""
        orchestrator = FixThenRevalidateOrchestrator(
            validation_pipeline=self.pipeline,
            max_retries=3,
        )
        failed = [_make_failed_result("bench-check")]
        orchestrator.run_fix_loop(
            plan_id="BENCH002",
            failed_results=failed,
            fix_callback=_always_fix,
            revalidate_callback=_CountingRevalidate(pass_after=2),
        )

    def time_terminal_failure(self) -> None:
        """Benchmark fix loop that results in terminal failure."""
        orchestrator = FixThenRevalidateOrchestrator(
            validation_pipeline=self.pipeline,
            max_retries=3,
        )
        failed = [_make_failed_result("bench-check")]
        orchestrator.run_fix_loop(
            plan_id="BENCH003",
            failed_results=failed,
            fix_callback=_always_fix,
            revalidate_callback=_NeverPassRevalidate(),
        )

    def time_empty_failures(self) -> None:
        """Benchmark fix loop with no failures (immediate pass)."""
        orchestrator = FixThenRevalidateOrchestrator(
            validation_pipeline=self.pipeline,
            max_retries=3,
        )
        orchestrator.run_fix_loop(
            plan_id="BENCH004",
            failed_results=[],
            fix_callback=_always_fix,
            revalidate_callback=_CountingRevalidate(pass_after=1),
        )


class FixRevalidateModelSuite:
    """Benchmarks for model creation and serialization."""

    timeout = 60

    def time_create_attempt_record(self) -> None:
        """Benchmark creating a FixAttemptRecord."""
        FixAttemptRecord(
            attempt_number=1,
            validation_name="bench-check",
            fix_description="Applied fix",
            success=True,
        )

    def time_create_result_model(self) -> None:
        """Benchmark creating a FixThenRevalidateResult."""
        records = [
            FixAttemptRecord(
                attempt_number=i + 1,
                validation_name=f"check-{i}",
                fix_description=f"Fix {i}",
                success=i == 2,
            )
            for i in range(3)
        ]
        FixThenRevalidateResult(
            validation_attempts=3,
            final_passed=True,
            validation_fix_history=records,
            escalated=False,
            terminal_failure=False,
        )

    def time_result_model_dump(self) -> None:
        """Benchmark serializing FixThenRevalidateResult to dict."""
        records = [
            FixAttemptRecord(
                attempt_number=1,
                validation_name="check-a",
                fix_description="Fix a",
                success=True,
            ),
        ]
        result = FixThenRevalidateResult(
            validation_attempts=1,
            final_passed=True,
            validation_fix_history=records,
        )
        result.model_dump()