cleveragents-core/docs/reference/event_bus.md

Event Bus

The EventBus infrastructure provides a general-purpose domain event system for CleverAgents. Domain services emit typed events at significant state changes (plan creation, decision recording, phase transitions, etc.), enabling audit logging, reactive subscribers, and observability without coupling services to one another.

---

Overview

The system consists of four cooperating components:

Component	Role
EventType	StrEnum of all domain event identifiers
DomainEvent	Frozen Pydantic model carrying event data
EventBus	Protocol that all bus implementations satisfy
ReactiveEventBus	RxPY-backed in-process bus (default)
LoggingEventBus	Structured-log-only bus (audit / test)

---

EventType

cleveragents.infrastructure.events.EventType is a StrEnum whose members cover all domains:

from cleveragents.infrastructure.events import EventType

print(EventType.PLAN_CREATED)        # "plan.created"
print(EventType.DECISION_CREATED)    # "decision.created"
print(EventType.ACTOR_COMPLETED)     # "actor.completed"

Categories:

Category	Examples
Plan lifecycle	plan.created, plan.phase_changed, plan.applied
Decision	decision.created, decision.superseded
Actor	actor.invoked, actor.completed, actor.errored
Tool	tool.invoked, tool.completed, tool.errored
Resource	resource.accessed, resource.modified
Sandbox	sandbox.created, checkpoint.restored
Validation	validation.started, validation.passed
Session	session.created, session.message_sent
Budget	budget.warning, budget.exceeded

---

DomainEvent

All events are instances of the frozen Pydantic model cleveragents.infrastructure.events.DomainEvent:

from cleveragents.infrastructure.events import DomainEvent, EventType

event = DomainEvent(
    event_type=EventType.PLAN_CREATED,
    plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
    actor_name="strategists/planner",
    details={"phase": "strategize"},
)

Field	Type	Default	Description
event_type	EventType	required	The type of event
timestamp	datetime	now(UTC)	UTC time of emission
correlation_id	str	auto ULID	Links related events in a request chain
plan_id	str | None	None	Associated plan ULID
root_plan_id	str | None	None	Root plan in a hierarchy
session_id	str | None	None	Associated session ULID
actor_name	str | None	None	Actor that emitted the event
project_name	str | None	None	Project context
details	dict	{}	Event-type-specific payload

DomainEvent is immutable (frozen=True). Attempting to set an attribute after construction raises a ValidationError.

---

EventBus Protocol

cleveragents.infrastructure.events.EventBus is a @runtime_checkable structural protocol. Any object with emit() and subscribe() methods satisfies it:

from cleveragents.infrastructure.events import EventBus, ReactiveEventBus

bus = ReactiveEventBus(max_audit_log_size=5000)  # optional retention cap
assert isinstance(bus, EventBus)  # True

Interface:

class EventBus(Protocol):
    def emit(self, event: DomainEvent) -> None: ...
    def subscribe(self, event_type: EventType,
                  handler: Callable[[DomainEvent], None]) -> None: ...

---

ReactiveEventBus

cleveragents.infrastructure.events.ReactiveEventBus is the default in-process implementation. It uses an RxPY Subject as its hot observable backbone.

from cleveragents.infrastructure.events import (
    DomainEvent, EventType, ReactiveEventBus,
)

bus = ReactiveEventBus()

# Subscribe a plain callback
bus.subscribe(EventType.PLAN_CREATED, lambda e: print("Plan created:", e.plan_id))

# Use the read-only RxPY stream for advanced operators
bus.stream.pipe(
    rx.operators.filter(lambda e: e.plan_id is not None),
).subscribe(lambda e: print("Plan event:", e))

# Emit an event
bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED, plan_id="01ABC..."))

• emit(event) pushes to the RxPY Subject and calls all type-specific handlers synchronously.
• subscribe(event_type, handler) registers a callback for one event type.
• stream exposes a read-only rx.Observable view for RxPY operators.
• clear_audit_log() clears retained in-memory events when needed.

---

LoggingEventBus

cleveragents.infrastructure.events.LoggingEventBus logs every event to structlog at INFO level. It is useful for audit trails, CI pipelines, and environments where RxPY overhead is undesirable.

from cleveragents.infrastructure.events import LoggingEventBus, EventType, DomainEvent

bus = LoggingEventBus()
bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED, plan_id="01ABC..."))
# → structured log: {"event": "domain_event", "event_type": "decision.created", ...}

---

Dependency Injection

The DI container (cleveragents.application.container.Container) registers ReactiveEventBus as a Singleton provider named event_bus. It is automatically wired into DecisionService and PlanLifecycleService:

from cleveragents.application.container import Container

container = Container()
bus = container.event_bus()   # ReactiveEventBus singleton

To substitute a different implementation (e.g. in tests):

from cleveragents.application.container import Container
from cleveragents.infrastructure.events import LoggingEventBus
from dependency_injector import providers

container = Container()
container.event_bus.override(providers.Singleton(LoggingEventBus))

---

Service Event Emission

DecisionService

cleveragents.application.services.decision_service.DecisionService.record_decision emits DECISION_CREATED after a decision is persisted.

PlanLifecycleService

Method	Event emitted
use_action(...)	PLAN_CREATED
execute_plan(...)	PLAN_PHASE_CHANGED

Both services accept an optional event_bus constructor parameter; when None, event emission is silently skipped (backward-compatible).

---

Example: Subscribing to Plan Events

from cleveragents.application.container import Container
from cleveragents.infrastructure.events import EventType, DomainEvent

container = Container()
bus = container.event_bus()

def on_plan_created(event: DomainEvent) -> None:
    print(f"New plan: {event.plan_id} | details: {event.details}")

bus.subscribe(EventType.PLAN_CREATED, on_plan_created)

# All subsequent plan.create operations will call on_plan_created
svc = container.plan_lifecycle_service()