cleveragents-core/benchmarks/plan_resume_bench.py

"""ASV benchmarks for plan resume overhead baseline.

Measures the cost of:
- Resume eligibility validation
- Step checkpoint recording
- Resume summary building (dry-run)
- Live resume state transitions
"""

from __future__ import annotations

from ulid import ULID

from cleveragents.application.services.plan_lifecycle_service import (
    PlanLifecycleService,
)
from cleveragents.application.services.plan_resume_service import (
    PlanResumeService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.plan import (
    ProjectLink,
)
from cleveragents.domain.models.core.resume import ResumeMetadata

_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
_bench_counter = 9000


def _bench_ulid() -> str:
    global _bench_counter
    _bench_counter += 1
    n = _bench_counter
    suffix = ""
    for _ in range(8):
        suffix = _CB32[n % 32] + suffix
        n //= 32
    return f"01HGZ6FE0AQDYTR4BX{suffix}"


def _make_services() -> tuple[PlanLifecycleService, PlanResumeService]:
    settings = Settings()
    lifecycle = PlanLifecycleService(settings=settings)
    resume = PlanResumeService(lifecycle_service=lifecycle)
    return lifecycle, resume


def _create_action(lifecycle: PlanLifecycleService, suffix: str) -> str:
    name = f"local/bench-resume-{suffix}"
    lifecycle.create_action(
        name=name,
        description="Benchmark resume action",
        definition_of_done="Step 1\nStep 2\nStep 3",
        strategy_actor="local/stub-strategy",
        execution_actor="local/stub-execute",
    )
    return name


def _make_errored_plan(lifecycle: PlanLifecycleService, action_name: str) -> str:
    plan = lifecycle.use_action(
        action_name=action_name,
        project_links=[ProjectLink(project_name="local/bench-proj")],
    )
    pid = plan.identity.plan_id
    lifecycle.start_strategize(pid)
    p = lifecycle.get_plan(pid)
    p.decision_root_id = str(ULID())
    lifecycle._commit_plan(p)
    lifecycle.complete_strategize(pid)
    lifecycle.execute_plan(pid)
    lifecycle.start_execute(pid)
    lifecycle.fail_execute(pid, "bench error")
    return pid


class TimeResumeEligibility:
    """Benchmark resume eligibility validation."""

    timeout = 30.0

    def setup(self) -> None:
        self.lifecycle, self.resume = _make_services()
        action = _create_action(self.lifecycle, _bench_ulid())
        self.plan_id = _make_errored_plan(self.lifecycle, action)

    def time_validate_eligibility(self) -> None:
        self.resume.validate_eligibility(self.plan_id)


class TimeCheckpointRecording:
    """Benchmark step checkpoint recording."""

    timeout = 30.0

    def setup(self) -> None:
        self.lifecycle, self.resume = _make_services()
        action = _create_action(self.lifecycle, _bench_ulid())
        plan = self.lifecycle.use_action(
            action_name=action,
            project_links=[ProjectLink(project_name="local/bench-proj")],
        )
        pid = plan.identity.plan_id
        self.lifecycle.start_strategize(pid)
        p = self.lifecycle.get_plan(pid)
        p.decision_root_id = str(ULID())
        self.lifecycle._commit_plan(p)
        self.lifecycle.complete_strategize(pid)
        self.lifecycle.execute_plan(pid)
        self.lifecycle.start_execute(pid)
        self.plan_id = pid
        self.resume.set_total_steps(pid, 100)
        self.step_counter = 0

    def time_record_checkpoint(self) -> None:
        idx = self.step_counter
        self.step_counter += 1
        if idx >= 100:
            return
        self.resume.record_step_checkpoint(
            self.plan_id, idx, f"DEC-{idx}", f"Step {idx}"
        )


class TimeDryRunResume:
    """Benchmark dry-run resume summary building."""

    timeout = 30.0

    def setup(self) -> None:
        self.lifecycle, self.resume = _make_services()
        action = _create_action(self.lifecycle, _bench_ulid())
        self.plan_id = _make_errored_plan(self.lifecycle, action)
        self.resume.set_total_steps(self.plan_id, 10)
        self.resume.record_step_checkpoint(self.plan_id, 5, "DEC-5", "Step 5")

    def time_dry_run_resume(self) -> None:
        self.resume.resume_plan(self.plan_id, dry_run=True)


class TimeResumeMetadataProperties:
    """Benchmark ResumeMetadata property access."""

    timeout = 30.0

    def setup(self) -> None:
        self.metadata = ResumeMetadata(
            last_completed_step=42,
            total_steps=100,
        )

    def time_has_progress(self) -> None:
        _ = self.metadata.has_progress

    def time_next_step_index(self) -> None:
        _ = self.metadata.next_step_index

    def time_is_complete(self) -> None:
        _ = self.metadata.is_complete