cleveragents-core/benchmarks/async_execution_bench.py

"""ASV benchmarks for async command execution and workers.

Measures the performance of:
- AsyncJob model creation and validation overhead
- AsyncJob state machine transitions
- InMemoryJobStore CRUD operations
- AsyncWorkerConfig instantiation
- WorkerHealthReport lifecycle
- CancellationToken operations
- Payload serialization/deserialization
- AsyncWorker job pickup and execution
"""

from __future__ import annotations

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

# Force-reload so ASV picks up the source tree version.
import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.application.services.async_worker import (  # noqa: E402
    AsyncWorker,
    AsyncWorkerConfig,
    CancellationToken,
    InMemoryJobStore,
    WorkerHealthReport,
)
from cleveragents.domain.models.core.async_job import (  # noqa: E402
    AsyncJob,
    AsyncJobStatus,
    InvalidJobTransitionError,
    can_transition_job,
    deserialize_job_payload,
    serialize_job_payload,
)

# A valid ULID for benchmarks
_ULID_A = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
_ULID_B = "01ARZ3NDEKTSV4RRFFQ69G5FAW"


# ---------------------------------------------------------------------------
# AsyncJob model creation benchmarks
# ---------------------------------------------------------------------------


class AsyncJobCreationSuite:
    """Benchmark AsyncJob model creation overhead."""

    timeout = 60

    def time_create_minimal_job(self) -> None:
        AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute")

    def time_create_job_with_payload(self) -> None:
        AsyncJob(
            job_id=_ULID_A,
            plan_id=_ULID_B,
            phase="apply",
            payload_json='{"key": "value", "count": 42}',
        )

    def time_create_job_all_fields(self) -> None:
        AsyncJob(
            job_id=_ULID_A,
            plan_id=_ULID_B,
            phase="execute",
            status=AsyncJobStatus.QUEUED,
            payload_json='{"schema_version": 1}',
            schema_version=1,
        )


# ---------------------------------------------------------------------------
# AsyncJob state machine benchmarks
# ---------------------------------------------------------------------------


class AsyncJobTransitionSuite:
    """Benchmark AsyncJob state machine transition overhead."""

    timeout = 60

    def setup(self) -> None:
        self._job_id = _ULID_A
        self._plan_id = _ULID_B

    def time_transition_queued_to_running(self) -> None:
        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
        job.mark_running("worker-bench")

    def time_transition_running_to_succeeded(self) -> None:
        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
        job.mark_running("worker-bench")
        job.mark_succeeded()

    def time_transition_running_to_failed(self) -> None:
        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
        job.mark_running("worker-bench")
        job.mark_failed()

    def time_transition_queued_to_cancelled(self) -> None:
        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
        job.mark_cancelled()

    def time_can_transition_check(self) -> None:
        can_transition_job(AsyncJobStatus.QUEUED, AsyncJobStatus.RUNNING)

    def time_full_lifecycle(self) -> None:
        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
        job.mark_running("worker-bench")
        job.record_heartbeat()
        job.mark_succeeded()

    def time_as_cli_dict(self) -> None:
        job = AsyncJob(job_id=self._job_id, plan_id=self._plan_id, phase="execute")
        job.as_cli_dict()

    def time_is_terminal_check(self) -> None:
        job = AsyncJob(
            job_id=self._job_id,
            plan_id=self._plan_id,
            phase="execute",
            status=AsyncJobStatus.QUEUED,
        )
        _ = job.is_terminal


# ---------------------------------------------------------------------------
# Payload serialization benchmarks
# ---------------------------------------------------------------------------


class PayloadSerializationSuite:
    """Benchmark payload serialization/deserialization overhead."""

    timeout = 60

    def time_serialize_minimal(self) -> None:
        serialize_job_payload(_ULID_A, "execute")

    def time_serialize_with_extra(self) -> None:
        serialize_job_payload(
            _ULID_A, "apply", extra={"steps": ["s1", "s2"], "retries": 3}
        )

    def time_deserialize_minimal(self) -> None:
        deserialize_job_payload('{"plan_id": "test", "phase": "execute"}')

    def time_deserialize_with_extra(self) -> None:
        deserialize_job_payload(
            '{"plan_id": "test", "phase": "apply", "extra": {"steps": ["s1"]}}'
        )

    def time_roundtrip(self) -> None:
        payload = serialize_job_payload(_ULID_A, "execute", extra={"key": "val"})
        deserialize_job_payload(payload)


# ---------------------------------------------------------------------------
# InMemoryJobStore benchmarks
# ---------------------------------------------------------------------------


class InMemoryJobStoreSuite:
    """Benchmark InMemoryJobStore CRUD operations."""

    timeout = 60

    def setup(self) -> None:
        self.store = InMemoryJobStore()
        # Pre-populate with 100 jobs
        for i in range(100):
            ulid = f"01ARZ3NDEKTSV4RRFFQ69G{i:04d}"[:26]
            job = AsyncJob(job_id=ulid, plan_id=_ULID_B, phase="execute")
            self.store.add(job)

    def time_add_job(self) -> None:
        store = InMemoryJobStore()
        job = AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute")
        store.add(job)

    def time_get_job(self) -> None:
        self.store.get(_ULID_A[:26])

    def time_list_by_status(self) -> None:
        self.store.list_by_status(AsyncJobStatus.QUEUED)

    def time_list_all(self) -> None:
        self.store.list_all()

    def time_count(self) -> None:
        self.store.count()


# ---------------------------------------------------------------------------
# AsyncWorkerConfig benchmarks
# ---------------------------------------------------------------------------


class AsyncWorkerConfigSuite:
    """Benchmark AsyncWorkerConfig instantiation overhead."""

    timeout = 60

    def time_default_config(self) -> None:
        AsyncWorkerConfig()

    def time_custom_config(self) -> None:
        AsyncWorkerConfig(
            enabled=True,
            max_workers=8,
            poll_interval=0.5,
            job_timeout=7200,
            job_ttl=172800,
        )


# ---------------------------------------------------------------------------
# WorkerHealthReport benchmarks
# ---------------------------------------------------------------------------


class WorkerHealthReportSuite:
    """Benchmark WorkerHealthReport lifecycle overhead."""

    timeout = 60

    def setup(self) -> None:
        self.report = WorkerHealthReport("worker-bench")

    def time_create_report(self) -> None:
        WorkerHealthReport("worker-bench")

    def time_record_heartbeat(self) -> None:
        self.report.record_heartbeat()

    def time_record_job_completed(self) -> None:
        self.report.record_job_completed()

    def time_record_job_failed(self) -> None:
        self.report.record_job_failed()

    def time_as_dict(self) -> None:
        self.report.as_dict()


# ---------------------------------------------------------------------------
# CancellationToken benchmarks
# ---------------------------------------------------------------------------


class CancellationTokenSuite:
    """Benchmark CancellationToken operations."""

    timeout = 60

    def time_create_token(self) -> None:
        CancellationToken()

    def time_check_not_cancelled(self) -> None:
        token = CancellationToken()
        _ = token.is_cancelled

    def time_cancel_and_check(self) -> None:
        token = CancellationToken()
        token.cancel()
        _ = token.is_cancelled

    def time_reset_token(self) -> None:
        token = CancellationToken()
        token.cancel()
        token.reset()

    def time_wait_with_timeout(self) -> None:
        token = CancellationToken()
        token.wait(timeout=0.0)


# ---------------------------------------------------------------------------
# AsyncWorker benchmarks
# ---------------------------------------------------------------------------


class AsyncWorkerSuite:
    """Benchmark AsyncWorker job execution overhead."""

    timeout = 60

    def setup(self) -> None:
        self.config = AsyncWorkerConfig(enabled=True)
        self.store = InMemoryJobStore()

    def time_create_worker(self) -> None:
        AsyncWorker(config=self.config, job_store=self.store)

    def time_pickup_and_execute(self) -> None:
        store = InMemoryJobStore()
        job = AsyncJob(job_id=_ULID_A, plan_id=_ULID_B, phase="execute")
        store.add(job)
        worker = AsyncWorker(config=self.config, job_store=store)
        worker.pickup_and_execute(job)

    def time_get_health_report(self) -> None:
        worker = AsyncWorker(config=self.config, job_store=self.store)
        worker.get_health_report()

    def time_detect_stuck_jobs_empty(self) -> None:
        worker = AsyncWorker(config=self.config, job_store=self.store)
        worker.detect_stuck_jobs()

    def time_cleanup_completed_jobs_empty(self) -> None:
        worker = AsyncWorker(config=self.config, job_store=self.store)
        worker.cleanup_completed_jobs()


# ---------------------------------------------------------------------------
# InvalidJobTransitionError benchmarks
# ---------------------------------------------------------------------------


class ErrorConstructionSuite:
    """Benchmark error construction overhead."""

    timeout = 60

    def time_invalid_transition_error(self) -> None:
        InvalidJobTransitionError(AsyncJobStatus.SUCCEEDED, AsyncJobStatus.RUNNING)