babylon.engine.event_bus

Event system for decoupled communication in the simulation.

This module provides a publish/subscribe event bus that enables loose coupling between simulation components. Events are immutable data objects that carry information about state changes.

Sprint 3: Central Committee (Dependency Injection) Epoch 1→2 Bridge: Added EventInterceptor pattern for adversarial mechanics.

Classes

Event(type, tick, payload[, timestamp])	Immutable event representing a simulation occurrence.
EventBus()	Publish/subscribe event bus for simulation components.

class babylon.engine.event_bus.Event(type, tick, payload, timestamp=<factory>)

Bases: object

Immutable event representing a simulation occurrence.

Events are frozen dataclasses to ensure they cannot be modified after creation, maintaining integrity of the event history.

Parameters:

• type (str)

• tick (int)

• payload (dict[str, Any])

• timestamp (datetime)

type

Event type identifier (e.g., “tick”, “rupture”, “synthesis”)

tick

Simulation tick when the event occurred

payload

Event-specific data dictionary

timestamp

Wall-clock time when event was created

type: str

tick: int

payload: dict[str, Any]

timestamp: datetime

__init__(type, tick, payload, timestamp=<factory>)

Parameters:

• type (str)

• tick (int)

• payload (dict[str, Any])

• timestamp (datetime)

Return type:

None

class babylon.engine.event_bus.EventBus

Bases: object

Publish/subscribe event bus for simulation components.

The EventBus enables decoupled communication between systems. Components can subscribe to specific event types and will be notified when events of that type are published.

All published events are stored in history for replay/debugging.

Epoch 1→2 Bridge: Supports optional interceptor chain for adversarial mechanics. If no interceptors are registered, events flow through with zero overhead (backwards compatible).

The interceptor chain processes events before emission: - Interceptors are sorted by priority (higher runs first) - Each interceptor can ALLOW, BLOCK, or MODIFY the event - If blocked, the event is logged and not emitted - If modified, the modified event continues through the chain

Example

>>> bus = EventBus()
>>> def on_tick(event: Event) -> None:
...     print(f"Tick {event.tick}: {event.payload}")
>>> bus.subscribe("tick", on_tick)
>>> bus.publish(Event(type="tick", tick=1, payload={"value": 42}))
Tick 1: {'value': 42}

__init__()

Initialize an empty event bus.

Return type:

None

subscribe(event_type, handler)

Subscribe a handler to receive events of a specific type.

Parameters:

• event_type (str) – The type of events to subscribe to

• handler (Callable[[Event], None]) – Callable that receives Event objects

Return type:

None

register_interceptor(interceptor)

Register an interceptor to process events before emission.

Interceptors are sorted by priority (higher first) each time an event is published. Multiple interceptors with the same priority execute in registration order.

Parameters:

interceptor (EventInterceptor) – The interceptor to register.

Return type:

None

Example

>>> from babylon.engine.interceptor import EventInterceptor
>>> bus = EventBus()
>>> bus.register_interceptor(my_security_interceptor)

unregister_interceptor(interceptor)

Remove an interceptor from the chain.

Parameters:

interceptor (EventInterceptor) – The interceptor to remove.

Raises:

ValueError – If the interceptor is not registered.

Return type:

None

publish(event, context=None)

Publish an event to all subscribed handlers.

If interceptors are registered, the event passes through the interceptor chain first. If any interceptor blocks the event, it is logged to the blocked events audit channel and not emitted.

The event is stored in history only if it passes all interceptors.

Parameters:

• event (Event) – The event to publish.

• context (WorldContext | None) – Optional world context for interceptors. Required for Epoch 2 adversarial mechanics.

Return type:

None

get_history()

Get a copy of all published events.

Return type:

list[Event]

Returns:

List of events in chronological order (oldest first).

get_blocked_events()

Get a copy of all blocked events.

The blocked events audit channel records every event that was stopped by an interceptor, including the blocking reason.

Return type:

list[BlockedEvent]

Returns:

List of BlockedEvent records in chronological order.

clear_history()

Remove all events from history.

Return type:

None

clear_blocked_events()

Remove all blocked event records.

Return type:

None

property interceptor_count: int

Number of registered interceptors.