"""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:
.. code-block:: text
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:
:class:`babylon.kernel.event_bus.Event`: The EventBus dataclass (internal)
:class:`babylon.models.world_state.WorldState`: Where events are stored
"""
from __future__ import annotations
from datetime import datetime
from typing import Annotated, Any, Literal
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, model_validator
from babylon.kernel.sim_clock import UNSET_TIMESTAMP, sim_datetime
from babylon.models.enums import EventType, GameOutcome
from babylon.models.types import Currency
[docs]
class SimulationEvent(BaseModel):
"""Base class for all simulation events (immutable).
All events share common fields for temporal tracking.
Subclasses add domain-specific fields.
Attributes:
event_type: The type of event (from EventType enum).
tick: Simulation tick when the event occurred (0-indexed).
timestamp: Deterministic sim-time derived from tick
(Constitution III.7).
Example:
Subclasses should set a default event_type::
class ExtractionEvent(EconomicEvent):
event_type: EventType = Field(default=EventType.SURPLUS_EXTRACTION)
"""
model_config = ConfigDict(frozen=True)
event_type: EventType = Field(
...,
description="Type of simulation event",
)
tick: int = Field(
ge=0,
description="Simulation tick when event occurred (0-indexed)",
)
timestamp: datetime = Field(
default=UNSET_TIMESTAMP,
description="Deterministic sim-time derived from tick (Constitution III.7)",
)
@model_validator(mode="before")
@classmethod
def _derive_timestamp_from_tick(cls, data: Any) -> Any:
"""III.7: default timestamps are a pure function of tick."""
if isinstance(data, dict):
ts = data.get("timestamp")
if ts is None or ts is UNSET_TIMESTAMP:
data = dict(data)
data["timestamp"] = sim_datetime(int(data.get("tick", 0)))
return data
[docs]
class EconomicEvent(SimulationEvent):
"""Economic events involving value transfer.
Base class for events that involve currency flow
(extraction, tribute, wages, subsidies).
Attributes:
amount: Currency amount involved in the transaction.
"""
amount: Currency = Field(
ge=0.0,
description="Currency amount involved in the transaction",
)
[docs]
class SubsidyEvent(EconomicEvent):
"""Imperial subsidy event (IMPERIAL_SUBSIDY).
Emitted when the core bourgeoisie subsidizes a client state to
maintain stability. Wealth converts to repression capacity.
Attributes:
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'>
"""
kind: Literal["imperial_subsidy"] = "imperial_subsidy"
event_type: EventType = Field(
default=EventType.IMPERIAL_SUBSIDY,
description="Event type (always IMPERIAL_SUBSIDY)",
)
source_id: str = Field(
...,
min_length=1,
description="Entity ID of the core bourgeoisie providing subsidy",
)
target_id: str = Field(
...,
min_length=1,
description="Entity ID of the client state receiving subsidy",
)
repression_boost: float = Field(
...,
ge=0.0,
description="Amount of repression capacity gained",
)
[docs]
class CrisisEvent(SimulationEvent):
"""Economic crisis event (ECONOMIC_CRISIS).
Emitted when the imperial rent pool depletes below critical threshold,
triggering bourgeoisie crisis response (wage cuts + repression).
Attributes:
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'>
"""
kind: Literal["economic_crisis"] = "economic_crisis"
event_type: EventType = Field(
default=EventType.ECONOMIC_CRISIS,
description="Event type (always ECONOMIC_CRISIS)",
)
pool_ratio: float = Field(
...,
ge=0.0,
description="Current pool divided by initial pool",
)
aggregate_tension: float = Field(
...,
ge=0.0,
le=1.0,
description="Average tension across all edges",
)
decision: str = Field(
...,
min_length=1,
description="Bourgeoisie decision (CRISIS, AUSTERITY, IRON_FIST, etc)",
)
wage_delta: float = Field(
...,
description="Change in wage rate (negative for cuts)",
)
# =============================================================================
# Carceral Equilibrium Events (Sprint 3.4+)
# =============================================================================
[docs]
class SuperwageCrisisEvent(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.
Attributes:
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'>
"""
kind: Literal["superwage_crisis"] = "superwage_crisis"
event_type: EventType = Field(
default=EventType.SUPERWAGE_CRISIS,
description="Event type (always SUPERWAGE_CRISIS)",
)
payer_id: str = Field(
...,
min_length=1,
description="Entity ID of the bourgeoisie who can't pay",
)
receiver_id: str = Field(
...,
min_length=1,
description="Entity ID of the labor aristocracy not receiving wages",
)
desired_wages: float = Field(
...,
ge=0.0,
description="Amount of wages that were needed",
)
available_pool: float = Field(
...,
description="Amount available in the rent pool",
)
[docs]
class ClassDecompositionEvent(SimulationEvent):
"""Class decomposition event (CLASS_DECOMPOSITION).
Emitted when the labor aristocracy splits into CARCERAL_ENFORCER
and INTERNAL_PROLETARIAT fractions after a super-wage crisis.
Attributes:
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,
... )
"""
kind: Literal["class_decomposition"] = "class_decomposition"
event_type: EventType = Field(
default=EventType.CLASS_DECOMPOSITION,
description="Event type (always CLASS_DECOMPOSITION)",
)
original_id: str = Field(
...,
min_length=1,
description="Entity ID of the labor aristocracy that split",
)
enforcer_fraction: float = Field(
default=0.3,
ge=0.0,
le=1.0,
description="Fraction that became enforcers",
)
proletariat_fraction: float = Field(
default=0.7,
ge=0.0,
le=1.0,
description="Fraction that became internal proletariat",
)
[docs]
class ControlRatioCrisisEvent(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.
Attributes:
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,
... )
"""
kind: Literal["control_ratio_crisis"] = "control_ratio_crisis"
event_type: EventType = Field(
default=EventType.CONTROL_RATIO_CRISIS,
description="Event type (always CONTROL_RATIO_CRISIS)",
)
prisoner_population: int = Field(
...,
ge=0,
description="Size of the prisoner/surplus population",
)
enforcer_population: int = Field(
...,
ge=0,
description="Size of the enforcer/guard population",
)
control_ratio: float = Field(
...,
ge=0.0,
description="Prisoners per enforcer",
)
capacity_threshold: float = Field(
...,
ge=0.0,
description="Maximum ratio enforcers can handle",
)
[docs]
class TerminalDecisionEvent(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.
Attributes:
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,
... )
"""
kind: Literal["terminal_decision"] = "terminal_decision"
event_type: EventType = Field(
default=EventType.TERMINAL_DECISION,
description="Event type (always TERMINAL_DECISION)",
)
outcome: str = Field(
...,
pattern="^(revolution|genocide)$",
description="Terminal outcome: revolution or genocide",
)
avg_organization: float = Field(
...,
ge=0.0,
le=1.0,
description="Average organization level of prisoners",
)
revolution_threshold: float = Field(
...,
ge=0.0,
le=1.0,
description="Threshold above which revolution occurs",
)
# =============================================================================
# Consciousness Events
# =============================================================================
[docs]
class ConsciousnessEvent(SimulationEvent):
"""Base class for consciousness-related events.
Events involving changes to class consciousness or ideological state.
Attributes:
target_id: Entity whose consciousness changed.
"""
target_id: str = Field(
...,
min_length=1,
description="Entity ID whose consciousness changed",
)
[docs]
class TransmissionEvent(ConsciousnessEvent):
"""Consciousness transmission event (CONSCIOUSNESS_TRANSMISSION).
Emitted when class consciousness flows from a revolutionary periphery
worker to a core worker via SOLIDARITY edges.
Attributes:
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'>
"""
kind: Literal["consciousness_transmission"] = "consciousness_transmission"
event_type: EventType = Field(
default=EventType.CONSCIOUSNESS_TRANSMISSION,
description="Event type (always CONSCIOUSNESS_TRANSMISSION)",
)
source_id: str = Field(
...,
min_length=1,
description="Entity ID transmitting consciousness",
)
delta: float = Field(
...,
description="Amount of consciousness transmitted",
)
solidarity_strength: float = Field(
...,
ge=0.0,
le=1.0,
description="Strength of the solidarity edge",
)
[docs]
class MassAwakeningEvent(ConsciousnessEvent):
"""Mass awakening event (MASS_AWAKENING).
Emitted when an entity's consciousness crosses the mass awakening
threshold, signifying a qualitative shift in class consciousness.
Attributes:
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'>
"""
kind: Literal["mass_awakening"] = "mass_awakening"
event_type: EventType = Field(
default=EventType.MASS_AWAKENING,
description="Event type (always MASS_AWAKENING)",
)
old_consciousness: float = Field(
...,
ge=0.0,
le=1.0,
description="Consciousness before awakening",
)
new_consciousness: float = Field(
...,
ge=0.0,
le=1.0,
description="Consciousness after awakening",
)
triggering_source: str = Field(
...,
min_length=1,
description="Entity ID that triggered the awakening",
)
# =============================================================================
# Struggle Events (Agency Layer - George Floyd Dynamic)
# =============================================================================
[docs]
class StruggleEvent(SimulationEvent):
"""Base class for struggle events (Agency Layer).
Events from the George Floyd Dynamic: Spark + Fuel = Explosion.
Attributes:
node_id: Entity where the struggle event occurred.
"""
node_id: str = Field(
...,
min_length=1,
description="Entity ID where struggle occurred",
)
[docs]
class SparkEvent(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.
Attributes:
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'>
"""
kind: Literal["excessive_force"] = "excessive_force"
event_type: EventType = Field(
default=EventType.EXCESSIVE_FORCE,
description="Event type (always EXCESSIVE_FORCE)",
)
repression: float = Field(
...,
ge=0.0,
le=1.0,
description="Current repression level faced by the entity",
)
spark_probability: float = Field(
...,
ge=0.0,
le=1.0,
description="Probability that led to this spark",
)
[docs]
class UprisingEvent(StruggleEvent):
"""Uprising event (UPRISING).
Emitted when a spark + accumulated agitation triggers mass insurrection.
The "explosion" in the George Floyd Dynamic.
Attributes:
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'>
"""
kind: Literal["uprising"] = "uprising"
event_type: EventType = Field(
default=EventType.UPRISING,
description="Event type (always UPRISING)",
)
trigger: str = Field(
...,
min_length=1,
description="What caused the uprising (spark or revolutionary_pressure)",
)
agitation: float = Field(
...,
ge=0.0,
description="Accumulated agitation level",
)
repression: float = Field(
...,
ge=0.0,
le=1.0,
description="Current repression level",
)
[docs]
class SolidaritySpikeEvent(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.
Attributes:
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'>
"""
kind: Literal["solidarity_spike"] = "solidarity_spike"
event_type: EventType = Field(
default=EventType.SOLIDARITY_SPIKE,
description="Event type (always SOLIDARITY_SPIKE)",
)
solidarity_gained: float = Field(
...,
ge=0.0,
description="Total solidarity strength gained",
)
edges_affected: int = Field(
...,
ge=0,
description="Number of solidarity edges strengthened",
)
triggered_by: str = Field(
...,
min_length=1,
description="What caused the spike (e.g., uprising)",
)
# =============================================================================
# Contradiction Events
# =============================================================================
[docs]
class ContradictionEvent(SimulationEvent):
"""Base class for dialectical contradiction events.
Events from tension dynamics and phase transitions.
Attributes:
edge: The edge where the contradiction occurred (format: "source->target").
"""
edge: str = Field(
...,
min_length=1,
description="Edge where contradiction occurred (format: source->target)",
)
[docs]
class RuptureEvent(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.
Attributes:
event_type: Always RUPTURE.
Example:
>>> event = RuptureEvent(
... tick=12,
... edge="C001->C002",
... )
>>> event.event_type
<EventType.RUPTURE: 'rupture'>
"""
kind: Literal["rupture"] = "rupture"
event_type: EventType = Field(
default=EventType.RUPTURE,
description="Event type (always RUPTURE)",
)
# Lawverian rewrite (Phase C1): RUPTURE is a FRAME-level event on the
# principal opposition, not a per-edge one. ``edge`` is relaxed to optional
# (no single edge ruptures now) and the opposition identity + gap/rate that
# tripped the condition-AND-level gate are carried instead.
edge: str = Field(
default="",
description="Deprecated (Phase C1): rupture is frame-level; empty by default.",
)
opposition: str = Field(
default="",
description="Key of the principal opposition that ruptured (e.g. capital_labor).",
)
gap: float = Field(
default=0.0,
ge=0.0,
le=1.0,
description="Principal opposition gap at rupture (> rupture_gap_threshold).",
)
rate: float = Field(
default=0.0,
description="Principal opposition gap rate at rupture (> 0: rising).",
)
# =============================================================================
# Topology Events (Sprint 3.3)
# =============================================================================
[docs]
class TopologyEvent(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.
Attributes:
percolation_ratio: Ratio of largest component to total nodes (L_max / N).
num_components: Number of disconnected solidarity components.
"""
percolation_ratio: float = Field(
ge=0.0,
le=1.0,
description="Ratio of largest component to total nodes",
)
num_components: int = Field(
ge=0,
description="Number of disconnected solidarity components",
)
[docs]
class PhaseTransitionEvent(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)
Attributes:
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'>
"""
kind: Literal["phase_transition"] = "phase_transition"
event_type: EventType = Field(
default=EventType.PHASE_TRANSITION,
description="Event type (always PHASE_TRANSITION)",
)
previous_state: str = Field(
...,
min_length=1,
description="Phase before transition (gaseous, transitional, liquid, solid)",
)
new_state: str = Field(
...,
min_length=1,
description="Phase after transition (gaseous, transitional, liquid, solid)",
)
largest_component_size: int = Field(
ge=0,
description="Size of the largest connected component (L_max)",
)
cadre_density: float = Field(
default=0.0,
ge=0.0,
le=1.0,
description="Ratio of cadre to sympathizers (actual/potential liquidity)",
)
is_resilient: bool | None = Field(
default=None,
description="Whether network survives purge (may be None if test not run)",
)
# =============================================================================
# Bifurcation Topology Events (Feature 033)
# =============================================================================
[docs]
class BifurcationTendencyEvent(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.
Attributes:
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.
"""
kind: Literal["bifurcation_tendency_change"] = "bifurcation_tendency_change"
event_type: EventType = Field(
default=EventType.BIFURCATION_TENDENCY_CHANGE,
description="Event type (always BIFURCATION_TENDENCY_CHANGE)",
)
previous_tendency: str = Field(
...,
min_length=1,
description="Overall tendency before change (revolutionary, fascist, indeterminate)",
)
new_tendency: str = Field(
...,
min_length=1,
description="Overall tendency after change (revolutionary, fascist, indeterminate)",
)
consciousness_weighted_cross_solidarity: float = Field(
ge=0.0,
description="Sum of consciousness-weighted cross-line solidarity",
)
mean_collective_identity_marginalized: float = Field(
ge=0.0,
le=1.0,
description="Mean CI across marginalized communities",
)
bridge_potential_weighted: float = Field(
ge=0.0,
description="Sum of infrastructure * sigmoid(CI) for bridges",
)
legitimation_index: float = Field(
ge=0.0,
le=1.0,
description="Population-weighted mean legitimation index",
)
# =============================================================================
# Endgame Events (Slice 1.6)
# =============================================================================
[docs]
class EndgameEvent(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
Attributes:
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'>
"""
kind: Literal["endgame_reached"] = "endgame_reached"
event_type: EventType = Field(
default=EventType.ENDGAME_REACHED,
description="Event type (always ENDGAME_REACHED)",
)
outcome: GameOutcome = Field(
...,
description="The game outcome that ended the simulation",
)
# =============================================================================
# Spec 057 — Leontief Imperial Rent Integration: CalibrationWarning event family
# =============================================================================
# These three events are infrastructural — they signal data-quality / calibration
# drift, not value transfer. Therefore they subclass SimulationEvent directly,
# NOT EconomicEvent (which mandates a Currency `amount`). The EconomicEvent
# placement originally proposed in research.md §R6 was revised at implementation
# time to honor the SimulationEvent vs EconomicEvent semantic distinction.
#
# Discriminator strings: "calibration_warning.<subtype>" per research.md §R6.
# Published via the existing EventBus.publish(Event(...)) adapter pattern.
# Subscribers filter by event.type.startswith("calibration_warning.") and
# parse payload back into the typed class via .model_validate(...).
[docs]
class AxiomViolationEvent(SimulationEvent):
"""Periphery-wage source published a ratio < 1.0 (FR-002).
Emitted by ``DefaultPeripheryLaborCoefficientsSource._fetch`` when the
structural axiom (core wages ≥ periphery wages, i.e. ratio ≥ 1.0) is
violated by source data. The source layer passes the value through
unchanged; the math layer (``ProductionChainRentCalculator``) clamps
via ``np.maximum(loss_ratio, 0.0)``. This event surfaces the
calibration signal without destabilizing downstream arithmetic
(research.md §R5 — two-layer pattern).
Attributes:
event_type: Always CALIBRATION_AXIOM_VIOLATION.
industry: BEA industry code where the violation occurred.
year: The data year.
ratio: The violating wage ratio (< threshold).
threshold: The expected lower bound (default 1.0).
"""
kind: Literal["calibration_warning.axiom_violation"] = "calibration_warning.axiom_violation"
event_type: EventType = Field(
default=EventType.CALIBRATION_AXIOM_VIOLATION,
description="Event type (always CALIBRATION_AXIOM_VIOLATION)",
)
industry: str = Field(..., min_length=1, description="BEA industry code")
year: int = Field(..., ge=1900, le=2100, description="Data year")
ratio: float = Field(..., description="The violating wage ratio (< threshold)")
threshold: float = Field(default=1.0, description="Expected lower bound (axiom)")
[docs]
class QcewCarryForwardEvent(SimulationEvent):
"""QCEW data missing for (county, year); employment shares carried forward (FR-004).
Emitted by ``DefaultIndustryToCountyAllocator.allocate`` when QCEW data
is missing for a (county, year) pair within the look-back window
(default 5 years per ``LeontiefRentDefines.qcew_carry_forward_max_years``).
Also emitted by ``imperial_rent.compute()`` with ``county_fips="*"`` and
``look_back_distance=-1`` as a sentinel for "Spec 057 pipeline not wired"
(graceful-degradation path per data-model.md ServiceContainer notes).
Attributes:
event_type: Always CALIBRATION_QCEW_CARRY_FORWARD.
county_fips: 5-char numeric FIPS (or "*" for "all counties" sentinel).
year: The tick year (gap year).
look_back_year: The year carried forward from.
look_back_distance: year - look_back_year (use -1 sentinel for
"Spec 057 pipeline not wired" pattern).
"""
kind: Literal["calibration_warning.qcew_carry_forward"] = (
"calibration_warning.qcew_carry_forward"
)
event_type: EventType = Field(
default=EventType.CALIBRATION_QCEW_CARRY_FORWARD,
description="Event type (always CALIBRATION_QCEW_CARRY_FORWARD)",
)
county_fips: str = Field(..., min_length=1, description="County FIPS or '*' sentinel")
year: int = Field(..., ge=1900, le=2100, description="Tick year (gap year)")
look_back_year: int = Field(..., ge=1900, le=2100, description="Year carried forward from")
look_back_distance: int = Field(
...,
ge=0,
le=20,
description="year - look_back_year (max 20; use the unsigned distance)",
)
[docs]
class PhiHourOutlierEvent(SimulationEvent):
"""Per-county phi_hour fell outside the LeontiefRentDefines plausibility bounds (FR-008).
Emitted by ``DefaultIndustryToCountyAllocator.allocate`` (or by
``imperial_rent.compute()`` post-allocation) when an allocated
``phi_hour`` falls outside ``[threshold_low, threshold_high]``. Defaults
come from ``LeontiefRentDefines.phi_hour_outlier_threshold_low/high``.
Attributes:
event_type: Always CALIBRATION_PHI_HOUR_OUTLIER.
county_fips: 5-char numeric FIPS where the outlier occurred.
phi_hour: The outlier value.
threshold_low: Plausibility lower bound (default -1000.0).
threshold_high: Plausibility upper bound (default 1000.0).
"""
kind: Literal["calibration_warning.phi_hour_outlier"] = "calibration_warning.phi_hour_outlier"
event_type: EventType = Field(
default=EventType.CALIBRATION_PHI_HOUR_OUTLIER,
description="Event type (always CALIBRATION_PHI_HOUR_OUTLIER)",
)
county_fips: str = Field(..., min_length=1, description="County FIPS where outlier occurred")
phi_hour: float = Field(..., description="The outlier value")
threshold_low: float = Field(default=-1000.0, description="Plausibility lower bound")
threshold_high: float = Field(default=1000.0, description="Plausibility upper bound")
# =============================================================================
# Event Deserialization (Sprint 1.X Deliverable 2)
# =============================================================================
# Registry mapping EventType values to their event classes
EVENT_CLASS_MAP: dict[str, type[SimulationEvent]] = {
EventType.SURPLUS_EXTRACTION.value: ExtractionEvent,
EventType.IMPERIAL_SUBSIDY.value: SubsidyEvent,
EventType.ECONOMIC_CRISIS.value: CrisisEvent,
EventType.SUPERWAGE_CRISIS.value: SuperwageCrisisEvent,
EventType.CLASS_DECOMPOSITION.value: ClassDecompositionEvent,
EventType.CONTROL_RATIO_CRISIS.value: ControlRatioCrisisEvent,
EventType.TERMINAL_DECISION.value: TerminalDecisionEvent,
EventType.CONSCIOUSNESS_TRANSMISSION.value: TransmissionEvent,
EventType.MASS_AWAKENING.value: MassAwakeningEvent,
EventType.EXCESSIVE_FORCE.value: SparkEvent,
EventType.UPRISING.value: UprisingEvent,
EventType.SOLIDARITY_SPIKE.value: SolidaritySpikeEvent,
EventType.RUPTURE.value: RuptureEvent,
EventType.PHASE_TRANSITION.value: PhaseTransitionEvent,
EventType.BIFURCATION_TENDENCY_CHANGE.value: BifurcationTendencyEvent,
EventType.ENDGAME_REACHED.value: EndgameEvent,
# Spec 057 — Leontief Imperial Rent Integration
EventType.CALIBRATION_AXIOM_VIOLATION.value: AxiomViolationEvent,
EventType.CALIBRATION_QCEW_CARRY_FORWARD.value: QcewCarryForwardEvent,
EventType.CALIBRATION_PHI_HOUR_OUTLIER.value: PhiHourOutlierEvent,
}
# Spec 059 US2 / FR-006 / SC-003: ``deserialize_event`` was DELETED in this
# bundle. All callers now use ``TickEventAdapter.validate_python(data)``
# directly. Legacy callers deserializing events without a ``kind`` field
# inject it from ``event_type`` first — see
# ``babylon.models.world_state._validate_event`` for the canonical pattern.
# ``EVENT_CLASS_MAP`` above is preserved as the kind→class lookup that
# ``_validate_event`` falls back on when neither ``kind`` nor ``event_type``
# discriminator is recoverable.
# =============================================================================
# TickEvent — Spec 059 US2 / ADR-004 (FR-004 + FR-005 + FR-006)
# =============================================================================
#
# TickEvent is the canonical Pydantic 2 *discriminated* union over the 19 leaf
# Event variants. Each leaf carries a unique ``kind: Literal["..."]`` field;
# Pydantic's ``Field(discriminator="kind")`` dispatches automatically on the
# kind value during validation.
#
# Status: the discriminator-field migration shipped in Spec 059 US2 (no
# longer deferred). ``deserialize_event`` now prefers ``TickEventAdapter``
# when ``kind`` is present in the input dict and falls back to the legacy
# ``EVENT_CLASS_MAP`` path only for older persisted data without a ``kind``
# field. This preserves backward compatibility with serialized world-states
# while enabling typecheck-time exhaustiveness for all new code.
#
# Each leaf variant retains its original ``event_type: EventType`` field
# (additive: both fields coexist on every instance). Removing ``event_type``
# is out-of-scope for Bundle 2 — it would break every test and call site
# that constructs an event with ``event_type=EventType.X`` as a keyword arg.
#
# Discriminated union (Spec 059 US2 / ADR-004): Pydantic dispatches on `kind`
TickEvent = Annotated[
ExtractionEvent
| SubsidyEvent
| CrisisEvent
| SuperwageCrisisEvent
| ClassDecompositionEvent
| ControlRatioCrisisEvent
| TerminalDecisionEvent
| TransmissionEvent
| MassAwakeningEvent
| SparkEvent
| UprisingEvent
| SolidaritySpikeEvent
| RuptureEvent
| PhaseTransitionEvent
| BifurcationTendencyEvent
| EndgameEvent
| AxiomViolationEvent
| QcewCarryForwardEvent
| PhiHourOutlierEvent,
Field(discriminator="kind"),
]
"""TickEvent: sum type for the 19 leaf Event variants. See block comment above."""
TickEventAdapter: TypeAdapter[TickEvent] = TypeAdapter(TickEvent)
"""TypeAdapter wrapping TickEvent for runtime validation against the Union."""