babylon.models.events

Pydantic event models for structured simulation events.

Sprint 3.1: Structured event persistence in WorldState. Sprint 3.1+: Expanded event type hierarchy for all 10 EventTypes.

These models replace raw dict payloads with typed, immutable event objects. Events are frozen Pydantic models that capture:

tick: When the event occurred
timestamp: Wall-clock time
event_type: EventType enum value
Additional type-specific fields

Design Principle: Events are IMMUTABLE FACTS about what happened. They should never be modified after creation.

Event Hierarchy:

SimulationEvent (base)
  |-- EconomicEvent (adds amount)
  |     |-- ExtractionEvent (SURPLUS_EXTRACTION)
  |     |-- SubsidyEvent (IMPERIAL_SUBSIDY)
  |     |-- CrisisEvent (ECONOMIC_CRISIS)
  |-- ConsciousnessEvent (adds target_id)
  |     |-- TransmissionEvent (CONSCIOUSNESS_TRANSMISSION)
  |     |-- MassAwakeningEvent (MASS_AWAKENING)
  |-- StruggleEvent (adds node_id)
  |     |-- SparkEvent (EXCESSIVE_FORCE)
  |     |-- UprisingEvent (UPRISING)
  |     |-- SolidaritySpikeEvent (SOLIDARITY_SPIKE)
  |-- ContradictionEvent (adds edge)
        |-- RuptureEvent (RUPTURE)

Usage:

from babylon.models.events import ExtractionEvent, UprisingEvent

event = ExtractionEvent(
tick=5, source_id=”C001”, target_id=”C002”, amount=10.5,

)

uprising = UprisingEvent(
tick=8, node_id=”C001”, trigger=”spark”, agitation=0.9, repression=0.7,

)

See also

babylon.engine.event_bus.Event: The EventBus dataclass (internal) babylon.models.world_state.WorldState: Where events are stored

Classes

`ConsciousnessEvent`(**data)	Base class for consciousness-related events.
`ContradictionEvent`(**data)	Base class for dialectical contradiction events.
`CrisisEvent`(**data)	Economic crisis event (ECONOMIC_CRISIS).
`EconomicEvent`(**data)	Economic events involving value transfer.
`ExtractionEvent`(**data)	Imperial rent extraction event (SURPLUS_EXTRACTION).
`MassAwakeningEvent`(**data)	Mass awakening event (MASS_AWAKENING).
`PhaseTransitionEvent`(**data)	Phase transition detected in solidarity network.
`RuptureEvent`(**data)	Rupture event (RUPTURE).
`SimulationEvent`(**data)	Base class for all simulation events (immutable).
`SolidaritySpikeEvent`(**data)	Solidarity spike event (SOLIDARITY_SPIKE).
`SparkEvent`(**data)	Excessive force spark event (EXCESSIVE_FORCE).
`StruggleEvent`(**data)	Base class for struggle events (Agency Layer).
`SubsidyEvent`(**data)	Imperial subsidy event (IMPERIAL_SUBSIDY).
`TopologyEvent`(**data)	Events related to network topology analysis.
`TransmissionEvent`(**data)	Consciousness transmission event (CONSCIOUSNESS_TRANSMISSION).
`UprisingEvent`(**data)	Uprising event (UPRISING).

class babylon.models.events.SimulationEvent(**data)[source]

Bases: BaseModel

Base class for all simulation events (immutable).

All events share common fields for temporal tracking. Subclasses add domain-specific fields.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)

event_type: The type of event (from EventType enum).

tick: Simulation tick when the event occurred (0-indexed).

timestamp: Wall-clock time when event was created.

Example

Subclasses should set a default event_type:

class ExtractionEvent(EconomicEvent):
    event_type: EventType = Field(default=EventType.SURPLUS_EXTRACTION)

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

event_type: EventType

tick: int

timestamp: datetime

class babylon.models.events.EconomicEvent(**data)[source]

Bases: SimulationEvent

Economic events involving value transfer.

Base class for events that involve currency flow (extraction, tribute, wages, subsidies).

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
amount (Annotated[float, Ge(ge=0.0), Ge(ge=0.0)])

amount: Currency amount involved in the transaction.

amount: Currency

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ExtractionEvent(**data)[source]

Bases: EconomicEvent

Imperial rent extraction event (SURPLUS_EXTRACTION).

Emitted when imperial rent is extracted from a periphery worker by the core bourgeoisie via EXPLOITATION edges.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
amount (Annotated[float, Ge(ge=0.0), Ge(ge=0.0)])
source_id (Annotated[str, MinLen(min_length=1)])
target_id (Annotated[str, MinLen(min_length=1)])
mechanism (str)

event_type: Always SURPLUS_EXTRACTION.

source_id: Entity ID of the worker being extracted from.

target_id: Entity ID of the bourgeoisie receiving rent.

mechanism: Description of extraction mechanism (default: “imperial_rent”).

Example

>>> event = ExtractionEvent(
...     tick=5,
...     source_id="C001",
...     target_id="C002",
...     amount=15.5,
... )
>>> event.event_type
<EventType.SURPLUS_EXTRACTION: 'surplus_extraction'>

event_type: EventType

source_id: str

target_id: str

mechanism: str

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.SubsidyEvent(**data)[source]

Bases: EconomicEvent

Imperial subsidy event (IMPERIAL_SUBSIDY).

Emitted when the core bourgeoisie subsidizes a client state to maintain stability. Wealth converts to repression capacity.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
amount (Annotated[float, Ge(ge=0.0), Ge(ge=0.0)])
source_id (Annotated[str, MinLen(min_length=1)])
target_id (Annotated[str, MinLen(min_length=1)])
repression_boost (Annotated[float, Ge(ge=0)])

event_type: Always IMPERIAL_SUBSIDY.

source_id: Entity ID of the core bourgeoisie providing subsidy.

target_id: Entity ID of the client state receiving subsidy.

repression_boost: Amount of repression capacity gained.

Example

>>> event = SubsidyEvent(
...     tick=5,
...     source_id="C002",
...     target_id="C003",
...     amount=100.0,
...     repression_boost=0.25,
... )
>>> event.event_type
<EventType.IMPERIAL_SUBSIDY: 'imperial_subsidy'>

event_type: EventType

source_id: str

target_id: str

repression_boost: float

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.CrisisEvent(**data)[source]

Bases: SimulationEvent

Economic crisis event (ECONOMIC_CRISIS).

Emitted when the imperial rent pool depletes below critical threshold, triggering bourgeoisie crisis response (wage cuts + repression).

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
pool_ratio (Annotated[float, Ge(ge=0)])
aggregate_tension (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
decision (Annotated[str, MinLen(min_length=1)])
wage_delta (float)

event_type: Always ECONOMIC_CRISIS.

pool_ratio: Current pool divided by initial pool.

aggregate_tension: Average tension across all edges.

decision: Bourgeoisie decision (CRISIS, AUSTERITY, IRON_FIST, etc).

wage_delta: Change in wage rate (negative for cuts).

Example

>>> event = CrisisEvent(
...     tick=10,
...     pool_ratio=0.15,
...     aggregate_tension=0.7,
...     decision="CRISIS",
...     wage_delta=-0.05,
... )
>>> event.event_type
<EventType.ECONOMIC_CRISIS: 'economic_crisis'>

event_type: EventType

pool_ratio: float

aggregate_tension: float

decision: str

wage_delta: float

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ConsciousnessEvent(**data)[source]

Bases: SimulationEvent

Base class for consciousness-related events.

Events involving changes to class consciousness or ideological state.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
target_id (Annotated[str, MinLen(min_length=1)])

target_id: Entity whose consciousness changed.

target_id: str

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.TransmissionEvent(**data)[source]

Bases: ConsciousnessEvent

Consciousness transmission event (CONSCIOUSNESS_TRANSMISSION).

Emitted when class consciousness flows from a revolutionary periphery worker to a core worker via SOLIDARITY edges.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
target_id (Annotated[str, MinLen(min_length=1)])
source_id (Annotated[str, MinLen(min_length=1)])
delta (float)
solidarity_strength (Annotated[float, Ge(ge=0.0), Le(le=1.0)])

event_type: Always CONSCIOUSNESS_TRANSMISSION.

source_id: Entity transmitting consciousness.

delta: Amount of consciousness transmitted.

solidarity_strength: Strength of the solidarity edge.

Example

>>> event = TransmissionEvent(
...     tick=3,
...     target_id="C001",
...     source_id="C002",
...     delta=0.05,
...     solidarity_strength=0.8,
... )
>>> event.event_type
<EventType.CONSCIOUSNESS_TRANSMISSION: 'consciousness_transmission'>

event_type: EventType

source_id: str

delta: float

solidarity_strength: float

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.MassAwakeningEvent(**data)[source]

Bases: ConsciousnessEvent

Mass awakening event (MASS_AWAKENING).

Emitted when an entity’s consciousness crosses the mass awakening threshold, signifying a qualitative shift in class consciousness.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
target_id (Annotated[str, MinLen(min_length=1)])
old_consciousness (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
new_consciousness (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
triggering_source (Annotated[str, MinLen(min_length=1)])

event_type: Always MASS_AWAKENING.

old_consciousness: Consciousness before awakening.

new_consciousness: Consciousness after awakening.

triggering_source: Entity that triggered the awakening.

Example

>>> event = MassAwakeningEvent(
...     tick=7,
...     target_id="C001",
...     old_consciousness=0.4,
...     new_consciousness=0.7,
...     triggering_source="C002",
... )
>>> event.event_type
<EventType.MASS_AWAKENING: 'mass_awakening'>

event_type: EventType

old_consciousness: float

new_consciousness: float

triggering_source: str

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.StruggleEvent(**data)[source]

Bases: SimulationEvent

Base class for struggle events (Agency Layer).

Events from the George Floyd Dynamic: Spark + Fuel = Explosion.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
node_id (Annotated[str, MinLen(min_length=1)])

node_id: Entity where the struggle event occurred.

node_id: str

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.SparkEvent(**data)[source]

Bases: StruggleEvent

Excessive force spark event (EXCESSIVE_FORCE).

Emitted when state violence (police brutality) occurs. This is the “spark” that can ignite an uprising if conditions are right.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
node_id (Annotated[str, MinLen(min_length=1)])
repression (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
spark_probability (Annotated[float, Ge(ge=0.0), Le(le=1.0)])

event_type: Always EXCESSIVE_FORCE.

repression: Current repression level faced by the entity.

spark_probability: Probability that led to this spark.

Example

>>> event = SparkEvent(
...     tick=5,
...     node_id="C001",
...     repression=0.8,
...     spark_probability=0.4,
... )
>>> event.event_type
<EventType.EXCESSIVE_FORCE: 'excessive_force'>

event_type: EventType

repression: float

spark_probability: float

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.UprisingEvent(**data)[source]

Bases: StruggleEvent

Uprising event (UPRISING).

Emitted when a spark + accumulated agitation triggers mass insurrection. The “explosion” in the George Floyd Dynamic.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
node_id (Annotated[str, MinLen(min_length=1)])
trigger (Annotated[str, MinLen(min_length=1)])
agitation (Annotated[float, Ge(ge=0)])
repression (Annotated[float, Ge(ge=0.0), Le(le=1.0)])

event_type: Always UPRISING.

trigger: What caused the uprising (“spark” or “revolutionary_pressure”).

agitation: Accumulated agitation level.

repression: Current repression level.

Example

>>> event = UprisingEvent(
...     tick=8,
...     node_id="C001",
...     trigger="spark",
...     agitation=0.9,
...     repression=0.7,
... )
>>> event.event_type
<EventType.UPRISING: 'uprising'>

event_type: EventType

trigger: str

agitation: float

repression: float

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.SolidaritySpikeEvent(**data)[source]

Bases: StruggleEvent

Solidarity spike event (SOLIDARITY_SPIKE).

Emitted when solidarity infrastructure is built through shared struggle. The lasting result of an uprising that enables future consciousness transmission.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
node_id (Annotated[str, MinLen(min_length=1)])
solidarity_gained (Annotated[float, Ge(ge=0)])
edges_affected (Annotated[int, Ge(ge=0)])
triggered_by (Annotated[str, MinLen(min_length=1)])

event_type: Always SOLIDARITY_SPIKE.

solidarity_gained: Total solidarity strength gained.

edges_affected: Number of solidarity edges strengthened.

triggered_by: What caused the spike (e.g., “uprising”).

Example

>>> event = SolidaritySpikeEvent(
...     tick=6,
...     node_id="C001",
...     solidarity_gained=0.3,
...     edges_affected=2,
...     triggered_by="uprising",
... )
>>> event.event_type
<EventType.SOLIDARITY_SPIKE: 'solidarity_spike'>

event_type: EventType

solidarity_gained: float

edges_affected: int

triggered_by: str

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.ContradictionEvent(**data)[source]

Bases: SimulationEvent

Base class for dialectical contradiction events.

Events from tension dynamics and phase transitions.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
edge (Annotated[str, MinLen(min_length=1)])

edge: The edge where the contradiction occurred (format: “source->target”).

edge: str

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.RuptureEvent(**data)[source]

Bases: ContradictionEvent

Rupture event (RUPTURE).

Emitted when tension on an edge reaches the critical threshold (1.0), triggering a phase transition. This represents the dialectical moment when accumulated contradictions become irreconcilable.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
edge (Annotated[str, MinLen(min_length=1)])

event_type: Always RUPTURE.

Example

>>> event = RuptureEvent(
...     tick=12,
...     edge="C001->C002",
... )
>>> event.event_type
<EventType.RUPTURE: 'rupture'>

event_type: EventType

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.TopologyEvent(**data)[source]

Bases: SimulationEvent

Events related to network topology analysis.

Base class for percolation theory metrics and phase transition detection. Tracks the state of the solidarity network structure.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
percolation_ratio (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
num_components (Annotated[int, Ge(ge=0)])

percolation_ratio: Ratio of largest component to total nodes (L_max / N).

num_components: Number of disconnected solidarity components.

percolation_ratio: float

num_components: int

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class babylon.models.events.PhaseTransitionEvent(**data)[source]

Bases: TopologyEvent

Phase transition detected in solidarity network.

Emitted when percolation_ratio crosses threshold boundaries.

4-Phase Model:

Gaseous (ratio < 0.1): Atomized, no coordination
Transitional (0.1 <= ratio < 0.5): Emerging structure
Liquid (ratio >= 0.5, cadre_density < 0.5): Mass movement (weak ties)
Solid (ratio >= 0.5, cadre_density >= 0.5): Vanguard party (strong ties)

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
percolation_ratio (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
num_components (Annotated[int, Ge(ge=0)])
previous_state (Annotated[str, MinLen(min_length=1)])
new_state (Annotated[str, MinLen(min_length=1)])
largest_component_size (Annotated[int, Ge(ge=0)])
cadre_density (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
is_resilient (bool | None)

event_type: Always PHASE_TRANSITION.

previous_state: Phase before transition (“gaseous”, “transitional”, “liquid”, “solid”).

new_state: Phase after transition.

largest_component_size: Size of the giant component (L_max).

cadre_density: Ratio of cadre to sympathizers (actual/potential liquidity).

is_resilient: Whether network survives 20% node removal (Sword of Damocles test).

Example

>>> event = PhaseTransitionEvent(
...     tick=10,
...     previous_state="gaseous",
...     new_state="liquid",
...     percolation_ratio=0.6,
...     num_components=2,
...     largest_component_size=12,
... )
>>> event.event_type
<EventType.PHASE_TRANSITION: 'phase_transition'>

event_type: EventType

previous_state: str

new_state: str

largest_component_size: int

cadre_density: float

is_resilient: bool | None

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].