cleveragents-core/benchmarks/plan_preflight_guardrails_bench.py

"""ASV benchmarks for plan pre-flight guardrail execution time.

Measures the performance of:
- Individual pre-flight checks
- Full ``run_all_checks`` pass
- ``PreflightReport`` construction
"""

from __future__ import annotations

import sys
from pathlib import Path

try:
    from cleveragents.application.services.plan_preflight_guardrail import (
        PlanPreflightGuardrail,
        PreflightCheckName,
        PreflightCheckResult,
        PreflightReport,
    )
except ModuleNotFoundError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
    from cleveragents.application.services.plan_preflight_guardrail import (
        PlanPreflightGuardrail,
        PreflightCheckName,
        PreflightCheckResult,
        PreflightReport,
    )


class PreflightCheckSuite:
    """Benchmark individual pre-flight check methods."""

    def setup(self) -> None:
        """Set up shared guardrail instance."""
        self.guardrail = PlanPreflightGuardrail()
        self.action_registry: dict[str, object] = {
            "test-action": {"name": "test-action"}
        }
        self.actor_registry: dict[str, object] = {
            "strategy": "s",
            "execution": "e",
            "estimation": "est",
            "invariant_reconciliation": "ir",
        }
        self.tool_registry: dict[str, object] = {
            "tool-a": {},
            "tool-b": {},
        }

    def time_check_action_schema(self) -> None:
        """Benchmark action schema validation check."""
        self.guardrail.check_action_schema("test-action", self.action_registry)

    def time_check_actor_availability(self) -> None:
        """Benchmark actor availability check."""
        self.guardrail.check_actor_availability(self.actor_registry)

    def time_check_skill_tool_existence(self) -> None:
        """Benchmark skill/tool existence check."""
        self.guardrail.check_skill_tool_existence(
            ("tool-a", "tool-b"), self.tool_registry, (), {}
        )

    def time_check_automation_policy(self) -> None:
        """Benchmark automation policy check."""
        self.guardrail.check_automation_policy({"name": "auto"})

    def time_check_rollback_feasibility_skip(self) -> None:
        """Benchmark rollback feasibility when checkpoints not required."""
        self.guardrail.check_rollback_feasibility(False, {})

    def time_check_rollback_feasibility_pass(self) -> None:
        """Benchmark rollback feasibility when all tools checkpointable."""
        self.guardrail.check_rollback_feasibility(
            True, {"tool-a": True, "tool-b": True}
        )

    def time_check_resource_accessibility_empty(self) -> None:
        """Benchmark resource accessibility with no paths."""
        self.guardrail.check_resource_accessibility(())

    def time_check_validation_resolution_empty(self) -> None:
        """Benchmark validation resolution with no validations."""
        self.guardrail.check_validation_resolution((), {})


class PreflightRunAllSuite:
    """Benchmark the full ``run_all_checks`` method."""

    def setup(self) -> None:
        """Set up shared guardrail with fully populated registries."""
        self.guardrail = PlanPreflightGuardrail()

    def time_run_all_checks_pass(self) -> None:
        """Benchmark run_all_checks with all checks passing."""
        self.guardrail.run_all_checks(
            action_name="test-action",
            action_registry={"test-action": {}},
            actor_registry={
                "strategy": "s",
                "execution": "e",
                "estimation": "est",
                "invariant_reconciliation": "ir",
            },
            tool_names=("tool-a",),
            tool_registry={"tool-a": {}},
            automation_profile={"name": "auto"},
            require_checkpoints=False,
        )


class PreflightReportSuite:
    """Benchmark PreflightReport construction and queries."""

    def time_report_construction(self) -> None:
        """Benchmark creating a report with 7 results."""
        report = PreflightReport()
        for check in PreflightCheckName:
            report.add(PreflightCheckResult(check, True, "ok"))

    def time_report_all_passed(self) -> None:
        """Benchmark all_passed property."""
        report = PreflightReport()
        for check in PreflightCheckName:
            report.add(PreflightCheckResult(check, True, "ok"))
        _ = report.all_passed

    def time_report_format(self) -> None:
        """Benchmark format_report rendering."""
        report = PreflightReport()
        for check in PreflightCheckName:
            report.add(PreflightCheckResult(check, True, "ok"))
        report.format_report()