babylon.engine.headless_runner.event_capture

Engine event capture for the headless runner.

Spec: 065-engine-bridging (T003 / T070).

The EventCapture subscribes to the engine’s EventBus and buffers every EventType fired during the tick loop. At end of run the buffer is drained and emitted into summary.json.events in deterministic emission order (FR-018).

This is the narrative spine of the run: every SuperwageCrisis, ClassDecomposition, ControlRatioCrisis, TerminalDecision, ExcessiveForce, Uprising, etc. fires through here.

Classes

EngineEvent(**data)	One fired engine event, awaiting serialization into summary.events.
EventCapture()	EventBus subscriber that collects engine events for summary.json.

class babylon.engine.headless_runner.event_capture.EngineEvent(**data)

Bases: BaseModel

One fired engine event, awaiting serialization into summary.events.

Parameters:

• tick (int)

• event_type (str)

• entity_ids (tuple[str, ...])

• severity (Literal['info', 'warning', 'error', 'critical'])

• details (dict[str, Any])

model_config: ClassVar[ConfigDict] = {'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tick: int

event_type: str

entity_ids: tuple[str, ...]

severity: Literal['info', 'warning', 'error', 'critical']

details: dict[str, Any]

class babylon.engine.headless_runner.event_capture.EventCapture

Bases: object

EventBus subscriber that collects engine events for summary.json.

__init__()

Return type:

None

property buffer_size: int

set_tick(tick)

Tag subsequent on_event calls with this tick number.

Called by the runner at the start of each tick BEFORE engine.run_tick(...).

Return type:

None

Parameters:

tick (int)

on_event(event)

EventBus subscriber callback. Preserves emission order.

Return type:

None

Parameters:

event (Any)

drain()

Return captured events (in emission order); buffer is preserved.

Idempotent calls return the same tuple (the buffer is not cleared on read — callers can re-drain for summary writing plus per-tick debug introspection).

Return type:

tuple[EngineEvent, ...]