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:

```python
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`:

```python
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:

```python
from cleveragents.infrastructure.events import EventBus, ReactiveEventBus

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

**Interface:**

```python
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.

```python
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.

```python
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`:

```python
from cleveragents.application.container import Container

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

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

```python
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

```python
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()
```