Engine

Core simulation engine and supporting components.

Simulation Engine

Simulation engine for the Babylon game loop.

The step() function is the core of Phase 2. It takes a WorldState and SimulationConfig and returns a new WorldState representing one tick of simulation time.

The step function is: - Pure: No side effects, no mutation of inputs - Deterministic: Same inputs always produce same outputs - Transparent: Order of operations encodes historical materialism

Turn Order (encodes historical materialism): 1. Economic Base - Value extraction (imperial rent) 2. Consciousness - Ideology drift based on material conditions 3. Survival Calculus - P(S|A) and P(S|R) updates 4. Contradiction Tension - Accumulated from wealth gaps 5. Event Logging - Record significant state changes

Phase 2.1: Refactored to modular System architecture. Phase 4a: Refactored to use ServiceContainer for dependency injection.

class babylon.engine.simulation_engine.SimulationEngine(systems)

Bases: object

Modular engine that advances the simulation by iterating through Systems.

The engine holds a list of systems and executes them in sequence. Order encodes historical materialism: 1. Economic Base (imperial rent) 2. Consciousness (ideology drift) 3. Survival Calculus (probability updates) 4. Contradiction (tension dynamics)

Parameters:

systems (list[System])

__init__(systems)

Initialize the engine with a list of systems.

Parameters:

systems (list[System]) – Ordered list of systems to execute each tick. Order matters! Economic systems must run before ideology.

Return type:

None

property systems: list[System]

Read-only access to registered systems.

run_tick(graph, services, context)

Execute all systems in order for one tick.

Parameters:

• graph – NetworkX graph (mutated in place by systems)

• services – ServiceContainer with config, formulas, event_bus, database

• context – TickContext or dict passed to all systems

babylon.engine.simulation_engine.step(state, config, persistent_context=None, defines=None)

Advance simulation by one tick using the modular engine.

This is the heart of Phase 2. It transforms a WorldState through one tick of simulated time by applying the MLM-TW formulas.

Parameters:

• state (WorldState) – Current world state (immutable)

• config (SimulationConfig) – Simulation configuration with formula coefficients

• persistent_context (dict[str, Any] | None) – Optional context dict that persists across ticks. Used by systems that need to track state between ticks (e.g., ConsciousnessSystem’s previous_wages for bifurcation mechanic).

• defines (GameDefines | None) – Optional custom GameDefines. If None, loads from default defines.yaml location. Use this for scenario-specific calibration.

Return type:

WorldState

Returns:

New WorldState at tick + 1

Order encodes historical materialism:

1. Economic base (value extraction)

2. Consciousness (responds to material conditions)

3. Survival calculus (probability updates)

4. Contradictions (tension from all above)

5. Event capture (log significant changes)

Simulation

Simulation facade class for running multi-tick simulations.

This module provides a Simulation class that wraps the ServiceContainer and step() function, providing a convenient API for: - Running simulations over multiple ticks - Preserving history of all WorldState snapshots - Maintaining a persistent ServiceContainer across ticks - Observer pattern for AI narrative generation (Sprint 3.1)

Observer Pattern Integration (Sprint 3.1): - Observers are registered via constructor or add_observer() - Notifications occur AFTER state reconstruction (per design decision) - Observer errors are logged but don’t halt simulation (ADR003) - Lifecycle hooks: on_simulation_start, on_tick, on_simulation_end

class babylon.engine.simulation.Simulation(initial_state, config, observers=None, defines=None)

Bases: object

Facade class for running multi-tick simulations with history preservation.

The Simulation class provides a stateful wrapper around the pure step() function, managing: - Current WorldState - History of all previous states - Persistent ServiceContainer for dependency injection - Observer notifications for AI/narrative components (Sprint 3.1)

Example

>>> from babylon.engine.factories import create_proletariat, create_bourgeoisie
>>> from babylon.models import WorldState, SimulationConfig, Relationship, EdgeType
>>>
>>> worker = create_proletariat()
>>> owner = create_bourgeoisie()
>>> exploitation = Relationship(
...     source_id=worker.id, target_id=owner.id,
...     edge_type=EdgeType.EXPLOITATION
... )
>>> state = WorldState(entities={worker.id: worker, owner.id: owner},
...                    relationships=[exploitation])
>>> config = SimulationConfig()
>>>
>>> sim = Simulation(state, config)
>>> sim.run(100)
>>> print(f"Worker wealth after 100 ticks: {sim.current_state.entities[worker.id].wealth}")

With observers:

>>> from babylon.ai import NarrativeDirector
>>> director = NarrativeDirector()
>>> sim = Simulation(state, config, observers=[director])
>>> sim.run(10)
>>> sim.end()  # Triggers on_simulation_end

Parameters:

• initial_state (WorldState)

• config (SimulationConfig)

• observers (list[SimulationObserver] | None)

• defines (GameDefines | None)

__init__(initial_state, config, observers=None, defines=None)

Initialize simulation with initial state and configuration.

Parameters:

• initial_state (WorldState) – Starting WorldState at tick 0

• config (SimulationConfig) – Simulation configuration with formula coefficients

• observers (list[SimulationObserver] | None) – Optional list of SimulationObserver instances to notify

• defines (GameDefines | None) – Optional custom GameDefines for scenario-specific coefficients. If None, loads from default defines.yaml location.

Return type:

None

property config: SimulationConfig

Return the simulation configuration.

property defines: GameDefines

Return the game defines.

property services: ServiceContainer

Return the persistent ServiceContainer.

property current_state: WorldState

Return the current WorldState.

property observers: list[SimulationObserver]

Return copy of registered observers.

Returns a copy to preserve encapsulation - modifying the returned list does not affect the internal observer list.

Returns:

A copy of the list of registered observers.

add_observer(observer)

Register an observer for simulation notifications.

Observers added after simulation has started will not receive on_simulation_start, but will receive on_tick and on_simulation_end notifications.

Parameters:

observer (SimulationObserver) – Observer implementing SimulationObserver protocol.

Return type:

None

remove_observer(observer)

Remove an observer. No-op if observer not present.

Parameters:

observer (SimulationObserver) – Observer to remove from notifications.

Return type:

None

step()

Advance simulation by one tick.

Applies the step() function to transform the current state, records the new state in history, updates current_state, and notifies registered observers.

On first step, observers receive on_simulation_start before on_tick.

The persistent context is passed to step() to maintain state across ticks (e.g., previous_wages for bifurcation mechanic).

Return type:

WorldState

Returns:

The new WorldState after one tick of simulation.

run(ticks)

Run simulation for N ticks.

Parameters:

ticks (int) – Number of ticks to advance the simulation

Return type:

WorldState

Returns:

The final WorldState after all ticks complete.

Raises:

ValueError – If ticks is negative

get_history()

Return all WorldState snapshots from initial to current.

The history includes: - Index 0: Initial state (tick 0) - Index N: State after N steps (tick N)

Return type:

list[WorldState]

Returns:

List of WorldState snapshots in chronological order.

update_state(new_state)

Update the current state mid-simulation.

This allows modifying the simulation state (e.g., changing relationships) while preserving the persistent context across ticks. Useful for testing scenarios like wage cuts where the previous_wages context must be preserved.

Parameters:

new_state (WorldState) – New WorldState to use as current state. The tick should match the expected continuation tick.

Return type:

None

Note

This does NOT add the new state to history - history reflects actual simulation progression, not manual state updates.

end()

Signal simulation end and notify observers.

Calls on_simulation_end on all registered observers with the current (final) state.

No-op if simulation has not started (no step() calls made). Can be called multiple times, but only the first call after step() will notify observers.

Return type:

None

Services

Service container for dependency injection.

This module provides a ServiceContainer dataclass that aggregates all dependencies needed by the simulation engine, enabling clean injection for testing and configuration.

Sprint 3: Central Committee (Dependency Injection) Paradox Refactor: Added GameDefines for centralized coefficients.

class babylon.engine.services.ServiceContainer(config, database, event_bus, formulas, defines)

Bases: object

Container for all simulation services.

Aggregates the five core services needed by the simulation: - config: Immutable simulation parameters - database: Database connection for persistence - event_bus: Publish/subscribe communication - formulas: Registry of mathematical formulas - defines: Centralized game coefficients (Paradox Refactor)

Example

>>> container = ServiceContainer.create()
>>> rent = container.formulas.get("imperial_rent")
>>> container.event_bus.publish(Event(...))
>>> with container.database.session() as session:
...     # do database work
>>> container.database.close()
>>> default_org = container.defines.DEFAULT_ORGANIZATION

Parameters:

• config (SimulationConfig)

• database (DatabaseConnection)

• event_bus (EventBus)

• formulas (FormulaRegistry)

• defines (GameDefines)

config: SimulationConfig

database: DatabaseConnection

event_bus: EventBus

formulas: FormulaRegistry

defines: GameDefines

classmethod create(config=None, defines=None)

Factory method to create a fully-initialized container.

Creates all services with sensible defaults. Uses in-memory SQLite for database isolation in tests.

Parameters:

• config (SimulationConfig | None) – Optional custom config. If None, uses default SimulationConfig.

• defines (GameDefines | None) – Optional custom defines. If None, uses default GameDefines.

Return type:

ServiceContainer

Returns:

ServiceContainer with all services initialized

__init__(config, database, event_bus, formulas, defines)

Parameters:

• config (SimulationConfig)

• database (DatabaseConnection)

• event_bus (EventBus)

• formulas (FormulaRegistry)

• defines (GameDefines)

Return type:

None

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.

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.

Formula Registry

Formula registry for hot-swappable mathematical functions.

This module provides a FormulaRegistry class that stores named callables, enabling runtime replacement of formulas for testing and modding.

Sprint 3: Central Committee (Dependency Injection)

class babylon.engine.formula_registry.FormulaRegistry

Bases: object

Registry for named mathematical formulas.

Provides a central lookup for all simulation formulas, enabling: - Hot-swapping formulas for testing with mocks - Modding support for custom formula implementations - Centralized formula management

Example

>>> registry = FormulaRegistry.default()
>>> rent = registry.get("imperial_rent")
>>> result = rent(alpha=0.5, periphery_wages=0.4, periphery_consciousness=0.2)

__init__()

Initialize an empty formula registry.

Return type:

None

register(name, func)

Register or replace a formula by name.

Parameters:

• name (str) – Unique identifier for the formula

• func (Callable[..., Any]) – Callable implementing the formula

Return type:

None

get(name)

Retrieve a formula by name.

Parameters:

name (str) – The formula identifier

Return type:

Callable[..., Any]

Returns:

The registered formula callable

Raises:

KeyError – If no formula is registered with the given name

list_formulas()

List all registered formula names.

Return type:

list[str]

Returns:

List of formula names in arbitrary order

classmethod default()

Create a registry pre-populated with all standard formulas.

Registers all 12 formulas from babylon.systems.formulas: - imperial_rent - labor_aristocracy_ratio - is_labor_aristocracy - consciousness_drift - acquiescence_probability - revolution_probability - crossover_threshold - loss_aversion - exchange_ratio - exploitation_rate - value_transfer - prebisch_singer

Return type:

FormulaRegistry

Returns:

FormulaRegistry with all standard formulas registered

Factories

Factory functions for creating simulation entities.

These functions provide convenient ways to create SocialClass entities with sensible defaults for class simulation. Each factory encapsulates the defaults appropriate for a specific social class.

Factories support the **kwargs pattern for extensibility while maintaining type safety through Pydantic validation.

Sprint 3.4.3 (George Jackson Refactor): ideology parameter accepts both float (legacy) and IdeologicalProfile (new format). Float values are automatically converted to IdeologicalProfile by the SocialClass validator.

babylon.engine.factories.create_proletariat(id='C001', name='Proletariat', wealth=0.5, ideology=-0.3, organization=0.1, repression_faced=0.5, subsistence_threshold=0.3, p_acquiescence=0.0, p_revolution=0.0, description='Exploited working class', effective_wealth=0.0, unearned_increment=0.0, ppp_multiplier=1.0)

Create a proletariat (exploited class) social class.

The proletariat is defined by: - PERIPHERY_PROLETARIAT role (exploited in the world system) - Low default wealth (0.5) - Slightly revolutionary ideology (-0.3) - Low organization (0.1 = 10%) - Moderate repression faced (0.5)

Parameters:

• id (str) – Unique identifier matching ^C[0-9]{3}$ pattern (default: “C001”)

• name (str) – Human-readable name (default: “Proletariat”)

• wealth (float) – Economic resources (default: 0.5)

• ideology (float | IdeologicalProfile) – Ideological position, -1=revolutionary to +1=reactionary (default: -0.3)

• organization (float) – Collective cohesion (default: 0.1)

• repression_faced (float) – State violence level (default: 0.5)

• subsistence_threshold (float) – Minimum wealth for survival (default: 0.3)

• p_acquiescence (float) – P(S|A) - survival through acquiescence (default: 0.0, calculated by engine)

• p_revolution (float) – P(S|R) - survival through revolution (default: 0.0, calculated by engine)

• description (str) – Optional description (default: “Exploited working class”)

• effective_wealth (float) – PPP-adjusted wealth (default: 0.0, calculated by engine)

• unearned_increment (float) – PPP bonus (default: 0.0, calculated by engine)

• ppp_multiplier (float) – PPP multiplier applied to wages (default: 1.0)

Return type:

SocialClass

Returns:

SocialClass configured as proletariat

Example

>>> worker = create_proletariat()
>>> worker.role
<SocialRole.PERIPHERY_PROLETARIAT: 'periphery_proletariat'>
>>> worker.wealth
0.5

babylon.engine.factories.create_bourgeoisie(id='C002', name='Bourgeoisie', wealth=10.0, ideology=0.8, organization=0.7, repression_faced=0.1, subsistence_threshold=0.1, p_acquiescence=0.0, p_revolution=0.0, description='Capital-owning exploiter class', effective_wealth=0.0, unearned_increment=0.0, ppp_multiplier=1.0)

Create a bourgeoisie (exploiter class) social class.

The bourgeoisie is defined by: - CORE_BOURGEOISIE role (exploiter in the world system) - High default wealth (10.0) - Reactionary ideology (0.8) - High organization (0.7 = 70%) - Low repression faced (0.1 - protected by state)

Parameters:

• id (str) – Unique identifier matching ^C[0-9]{3}$ pattern (default: “C002”)

• name (str) – Human-readable name (default: “Bourgeoisie”)

• wealth (float) – Economic resources (default: 10.0)

• ideology (float | IdeologicalProfile) – Ideological position, -1=revolutionary to +1=reactionary (default: 0.8)

• organization (float) – Collective cohesion (default: 0.7)

• repression_faced (float) – State violence level (default: 0.1)

• subsistence_threshold (float) – Minimum wealth for survival (default: 0.1)

• p_acquiescence (float) – P(S|A) - survival through acquiescence (default: 0.0, calculated by engine)

• p_revolution (float) – P(S|R) - survival through revolution (default: 0.0, calculated by engine)

• description (str) – Optional description (default: “Capital-owning exploiter class”)

• effective_wealth (float) – PPP-adjusted wealth (default: 0.0, calculated by engine)

• unearned_increment (float) – PPP bonus (default: 0.0, calculated by engine)

• ppp_multiplier (float) – PPP multiplier applied to wages (default: 1.0)

Return type:

SocialClass

Returns:

SocialClass configured as bourgeoisie

Example

>>> owner = create_bourgeoisie()
>>> owner.role
<SocialRole.CORE_BOURGEOISIE: 'core_bourgeoisie'>
>>> owner.wealth
10.0