babylon.engine.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

Classes

Simulation(initial_state, config[, ...])

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

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