aditya
d17895dae2
- Add EventType StrEnum with 38 typed domain event identifiers across 8 domains (plan lifecycle, decision, actor, tool, resource, sandbox, validation, session, budget) - Add frozen DomainEvent Pydantic model with event_type, auto-UTC timestamp, auto-ULID correlation_id, plan_id, root_plan_id, session_id, actor_name, project_name, and details fields - Add @runtime_checkable EventBus Protocol with emit() and subscribe() - Add ReactiveEventBus backed by RxPY Subject for in-process reactive streaming with synchronous handler dispatch and raw Observable stream - Add LoggingEventBus using structlog for audit-trail-only environments - Wire DecisionService to emit DECISION_CREATED on record_decision() - Wire PlanLifecycleService to emit PLAN_CREATED and PLAN_PHASE_CHANGED on use_action() and execute_plan() respectively - Register ReactiveEventBus as Singleton in DI container and inject into DecisionService and PlanLifecycleService providers - Add Behave BDD unit tests: 27 scenarios, 75 steps, 100% branch coverage - Add Robot Framework integration tests: 9 smoke cases - Add ASV performance benchmarks: 5 suites covering emit throughput and subscriber fan-out - Add reference documentation at docs/reference/event_bus.md ISSUES CLOSED: #473
160 lines
4.5 KiB
Python
160 lines
4.5 KiB
Python
"""ASV benchmarks for EventBus infrastructure.
|
|
|
|
Measures:
|
|
- DomainEvent construction throughput
|
|
- ReactiveEventBus emit overhead (no subscribers)
|
|
- ReactiveEventBus emit with one subscriber
|
|
- ReactiveEventBus subscribe registration
|
|
- LoggingEventBus emit overhead
|
|
- Fan-out: emit to N subscribers
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import ClassVar
|
|
|
|
_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.infrastructure.events import ( # noqa: E402
|
|
DomainEvent,
|
|
EventType,
|
|
LoggingEventBus,
|
|
ReactiveEventBus,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DomainEvent construction
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class DomainEventSuite:
|
|
"""Benchmark DomainEvent construction overhead."""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
self.event_type = EventType.PLAN_CREATED
|
|
self.plan_id = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
|
|
|
|
def time_create_minimal(self) -> None:
|
|
DomainEvent(event_type=self.event_type)
|
|
|
|
def time_create_with_plan_id(self) -> None:
|
|
DomainEvent(event_type=self.event_type, plan_id=self.plan_id)
|
|
|
|
def time_create_with_details(self) -> None:
|
|
DomainEvent(
|
|
event_type=self.event_type,
|
|
plan_id=self.plan_id,
|
|
details={"phase": "strategize", "actor": "strategists/planner"},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ReactiveEventBus emit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ReactiveEmitSuite:
|
|
"""Benchmark ReactiveEventBus.emit() throughput."""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
self.bus = ReactiveEventBus()
|
|
self.event = DomainEvent(event_type=EventType.PLAN_CREATED)
|
|
self.decision_event = DomainEvent(event_type=EventType.DECISION_CREATED)
|
|
|
|
# Pre-register a subscriber for the with-subscriber benchmarks
|
|
self._received: list[DomainEvent] = []
|
|
self.bus.subscribe(EventType.PLAN_CREATED, self._received.append)
|
|
|
|
def time_emit_no_subscribers(self) -> None:
|
|
self.bus.emit(self.decision_event)
|
|
|
|
def time_emit_with_one_subscriber(self) -> None:
|
|
self.bus.emit(self.event)
|
|
|
|
def time_emit_100_events(self) -> None:
|
|
for _ in range(100):
|
|
self.bus.emit(self.event)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ReactiveEventBus subscribe
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ReactiveSubscribeSuite:
|
|
"""Benchmark ReactiveEventBus.subscribe() overhead."""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
self.event_type = EventType.PLAN_CREATED
|
|
|
|
def time_subscribe_single(self) -> None:
|
|
bus = ReactiveEventBus()
|
|
bus.subscribe(self.event_type, lambda e: None)
|
|
|
|
def time_subscribe_10(self) -> None:
|
|
bus = ReactiveEventBus()
|
|
for _ in range(10):
|
|
bus.subscribe(self.event_type, lambda e: None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fan-out benchmark
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class FanOutSuite:
|
|
"""Benchmark event fan-out to multiple subscribers."""
|
|
|
|
timeout = 60
|
|
params: ClassVar[list[int]] = [1, 5, 10, 50]
|
|
param_names: ClassVar[list[str]] = ["num_subscribers"]
|
|
|
|
def setup(self, num_subscribers: int) -> None:
|
|
self.bus = ReactiveEventBus()
|
|
for _ in range(num_subscribers):
|
|
self.bus.subscribe(EventType.PLAN_CREATED, lambda e: None)
|
|
self.event = DomainEvent(event_type=EventType.PLAN_CREATED)
|
|
|
|
def time_emit(self, num_subscribers: int) -> None:
|
|
self.bus.emit(self.event)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LoggingEventBus emit
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class LoggingEmitSuite:
|
|
"""Benchmark LoggingEventBus.emit() overhead."""
|
|
|
|
timeout = 60
|
|
|
|
def setup(self) -> None:
|
|
import logging
|
|
|
|
logging.disable(logging.CRITICAL)
|
|
self.bus = LoggingEventBus()
|
|
self.event = DomainEvent(event_type=EventType.DECISION_CREATED)
|
|
|
|
def time_emit_single(self) -> None:
|
|
self.bus.emit(self.event)
|
|
|
|
def time_emit_100(self) -> None:
|
|
for _ in range(100):
|
|
self.bus.emit(self.event)
|