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

Functions

deserialize_event(data)

Deserialize an event from a dict to the appropriate SimulationEvent subclass.

Classes

`BifurcationTendencyEvent`(**data)	Bifurcation tendency change detected in solidarity network.
`ClassDecompositionEvent`(**data)	Class decomposition event (CLASS_DECOMPOSITION).
`ConsciousnessEvent`(**data)	Base class for consciousness-related events.
`ContradictionEvent`(**data)	Base class for dialectical contradiction events.
`ControlRatioCrisisEvent`(**data)	Control ratio crisis event (CONTROL_RATIO_CRISIS).
`CrisisEvent`(**data)	Economic crisis event (ECONOMIC_CRISIS).
`EconomicEvent`(**data)	Economic events involving value transfer.
`EndgameEvent`(**data)	Endgame reached event (ENDGAME_REACHED).
`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).
`SuperwageCrisisEvent`(**data)	Super-wage crisis event (SUPERWAGE_CRISIS).
`TerminalDecisionEvent`(**data)	Terminal decision event (TERMINAL_DECISION).
`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), AfterValidator(func=~babylon.utils.math.quantize)])

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), AfterValidator(func=~babylon.utils.math.quantize)])
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), AfterValidator(func=~babylon.utils.math.quantize)])
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.SuperwageCrisisEvent(**data)[source]

Bases: SimulationEvent

Super-wage crisis event (SUPERWAGE_CRISIS).

Emitted when the imperial rent pool is exhausted and core bourgeoisie can no longer afford to pay super-wages to the labor aristocracy. This triggers the Carceral Turn phase transition.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
payer_id (Annotated[str, MinLen(min_length=1)])
receiver_id (Annotated[str, MinLen(min_length=1)])
desired_wages (Annotated[float, Ge(ge=0)])
available_pool (float)

event_type: Always SUPERWAGE_CRISIS.

payer_id: Entity ID of the bourgeoisie who can’t pay.

receiver_id: Entity ID of the labor aristocracy not receiving wages.

desired_wages: Amount of wages that were needed.

available_pool: Amount available in the rent pool (zero or negative).

Example

>>> event = SuperwageCrisisEvent(
...     tick=1040,
...     payer_id="C003",
...     receiver_id="C004",
...     desired_wages=5.0,
...     available_pool=0.0,
... )
>>> event.event_type
<EventType.SUPERWAGE_CRISIS: 'superwage_crisis'>

event_type: EventType

payer_id: str

receiver_id: str

desired_wages: float

available_pool: 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.ClassDecompositionEvent(**data)[source]

Bases: SimulationEvent

Class decomposition event (CLASS_DECOMPOSITION).

Emitted when the labor aristocracy splits into CARCERAL_ENFORCER and INTERNAL_PROLETARIAT fractions after a super-wage crisis.

Parameters:

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

event_type: Always CLASS_DECOMPOSITION.

original_id: Entity ID of the labor aristocracy that split.

enforcer_fraction: Fraction that became enforcers (default 0.3).

proletariat_fraction: Fraction that became internal proletariat (0.7).

Example

>>> event = ClassDecompositionEvent(
...     tick=1092,
...     original_id="C004",
...     enforcer_fraction=0.3,
...     proletariat_fraction=0.7,
... )

event_type: EventType

original_id: str

enforcer_fraction: float

proletariat_fraction: 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.ControlRatioCrisisEvent(**data)[source]

Bases: SimulationEvent

Control ratio crisis event (CONTROL_RATIO_CRISIS).

Emitted when the prisoner-to-guard ratio exceeds capacity, meaning the carceral apparatus can no longer contain the surplus population.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
prisoner_population (Annotated[int, Ge(ge=0)])
enforcer_population (Annotated[int, Ge(ge=0)])
control_ratio (Annotated[float, Ge(ge=0)])
capacity_threshold (Annotated[float, Ge(ge=0)])

event_type: Always CONTROL_RATIO_CRISIS.

prisoner_population: Size of the prisoner/surplus population.

enforcer_population: Size of the enforcer/guard population.

control_ratio: Prisoners per enforcer.

capacity_threshold: Maximum ratio enforcers can handle.

Example

>>> event = ControlRatioCrisisEvent(
...     tick=2340,
...     prisoner_population=1000,
...     enforcer_population=100,
...     control_ratio=10.0,
...     capacity_threshold=5.0,
... )

event_type: EventType

prisoner_population: int

enforcer_population: int

control_ratio: float

capacity_threshold: 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.TerminalDecisionEvent(**data)[source]

Bases: SimulationEvent

Terminal decision event (TERMINAL_DECISION).

Emitted when the system bifurcates to either revolution or genocide based on the organization level of the surplus population.

Parameters:

event_type (EventType)
tick (Annotated[int, Ge(ge=0)])
timestamp (datetime)
outcome (Annotated[str, _PydanticGeneralMetadata(pattern='^(revolution|genocide)$')])
avg_organization (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
revolution_threshold (Annotated[float, Ge(ge=0.0), Le(le=1.0)])

event_type: Always TERMINAL_DECISION.

outcome: Either “revolution” or “genocide”.

avg_organization: Average organization level of prisoners.

revolution_threshold: Threshold above which revolution occurs.

Example

>>> event = TerminalDecisionEvent(
...     tick=2860,
...     outcome="revolution",
...     avg_organization=0.65,
...     revolution_threshold=0.6,
... )

event_type: EventType

outcome: str

avg_organization: float

revolution_threshold: 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].

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

Bases: TopologyEvent

Bifurcation tendency change detected in solidarity network.

Emitted when the overall bifurcation tendency (revolutionary/fascist/ indeterminate) changes between ticks. Consciousness-weighted analysis detects whether crisis routes to fascism or revolution.

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_tendency (Annotated[str, MinLen(min_length=1)])
new_tendency (Annotated[str, MinLen(min_length=1)])
consciousness_weighted_cross_solidarity (Annotated[float, Ge(ge=0)])
mean_collective_identity_marginalized (Annotated[float, Ge(ge=0.0), Le(le=1.0)])
bridge_potential_weighted (Annotated[float, Ge(ge=0)])
legitimation_index (Annotated[float, Ge(ge=0.0), Le(le=1.0)])

event_type: Always BIFURCATION_TENDENCY_CHANGE.

previous_tendency: Overall tendency before change.

new_tendency: Overall tendency after change.

consciousness_weighted_cross_solidarity: Sum of consciousness-weighted cross-line solidarity edges.

mean_collective_identity_marginalized: Mean CI across marginalized communities.

bridge_potential_weighted: Sum of infrastructure * sigmoid(CI) for communities bridging contradiction axes.

legitimation_index: Population-weighted mean legitimation index.

event_type: EventType

previous_tendency: str

new_tendency: str

consciousness_weighted_cross_solidarity: float

mean_collective_identity_marginalized: float

bridge_potential_weighted: float

legitimation_index: 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.EndgameEvent(**data)[source]

Bases: SimulationEvent

Endgame reached event (ENDGAME_REACHED).

Emitted when a game-ending condition is met. The simulation terminates after this event with the specified outcome.

Outcomes:

REVOLUTIONARY_VICTORY: Proletarian revolution succeeded
ECOLOGICAL_COLLAPSE: Metabolic rift has become fatal
FASCIST_CONSOLIDATION: Fascism has consolidated power

Parameters:

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

event_type: Always ENDGAME_REACHED.

outcome: The GameOutcome that ended the simulation.

Example

>>> event = EndgameEvent(
...     tick=50,
...     outcome=GameOutcome.REVOLUTIONARY_VICTORY,
... )
>>> event.event_type
<EventType.ENDGAME_REACHED: 'endgame_reached'>

event_type: EventType

outcome: GameOutcome

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

babylon.models.events.deserialize_event(data)[source]

Deserialize an event from a dict to the appropriate SimulationEvent subclass.

This is used by WorldState.from_graph() to restore events from graph metadata. The event_type field determines which subclass to use for deserialization.

Parameters:: data (dict[str, object]) – Dict containing event data, must have “event_type” key.
Return type:: SimulationEvent
Returns:: The appropriate SimulationEvent subclass instance.
Raises:: ValueError – If event_type is missing or unrecognized.

Example

>>> data = {"event_type": "surplus_extraction", "tick": 5, ...}
>>> event = deserialize_event(data)
>>> isinstance(event, ExtractionEvent)
True