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

MVP Simulation Engine (001-mvp-sim-engine): - Implements SimulationState and SimulationControl protocols - Tracks territory profit_rate for GUI visualization - Provides get_snapshot(), get_territory_state(), reset() methods - Deterministic profit_rate decay toward territory-specific equilibrium

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, tensor_registry=None, calculator_overrides=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)

• tensor_registry (TensorRegistry | None)

• calculator_overrides (dict[str, Any] | None)

__init__(initial_state, config, observers=None, defines=None, tensor_registry=None, calculator_overrides=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.

• tensor_registry (TensorRegistry | None) – Optional TensorRegistry for cached tensor data access. If None, tensor data is not available. If provided, it should be pre-hydrated with the relevant counties and years.

• calculator_overrides (dict[str, Any] | None) – Optional dict of calculator instances to inject into ServiceContainer on each tick (Feature 020).

Return type:

None

classmethod from_sqlite(fips_codes, year=2022, observers=None, defines=None, years=None)

Create simulation initialized from SQLite reference database.

This is the main entry point for the MVP simulation engine. It hydrates territories from the reference database with profit_rate computed from QCEW/BEA data.

Parameters:

• fips_codes (list[str]) – List of 5-digit FIPS codes for counties to simulate. Example: [“26163”, “26125”] for Wayne and Oakland counties.

• year (int) – Data year for QCEW/BEA data (default 2022).

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

• defines (GameDefines | None) – Optional custom GameDefines for scenario-specific coefficients.

• years (Sequence[int] | None) – Optional sequence of years for multi-year time series. When provided, tensor data is hydrated for all specified years and the economics calculator factory is wired automatically.

Return type:

Simulation

Returns:

Initialized Simulation with territories hydrated from database.

Raises:

ValueError – If fips_codes is empty, contains duplicates that reduce to fewer unique codes, or any county is not found in database.

Example

>>> sim = Simulation.from_sqlite(
...     fips_codes=["26163", "26125"],  # Detroit metro
...     year=2022
... )
>>> snapshot = sim.get_snapshot()
>>> wayne = snapshot.territories["26163"]
>>> print(f"Wayne County profit rate: {wayne.profit_rate}")

See also

• plan.md#Hydration Flow

• quickstart.md

property config: SimulationConfig

Return the simulation configuration.

property defines: GameDefines

Return the game defines.

property services: ServiceContainer

Return the persistent ServiceContainer.

property tensor_registry: TensorRegistry | None

Return the TensorRegistry for cached economic data access.

Returns:

TensorRegistry if initialized, None otherwise.

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

register_observer(callback)

Register a GUI callback for tick notifications.

Implements SimulationControl protocol.

Thread Safety:

Callbacks receive a frozen SimulationSnapshot, not a live reference to mutable simulation state. The ProtocolObserverAdapter creates the snapshot BEFORE iterating callbacks, ensuring: - All callbacks see the same consistent state - GUI code cannot race with engine mutations - Callback processing time does not affect snapshot consistency

Callbacks are invoked in registration order. Duplicate registration is idempotent (callback invoked once per tick).

Parameters:

callback (Callable[[int, SimulationSnapshot], None]) – Function to call after each tick. Signature: (tick: int, snapshot: SimulationSnapshot) -> None

Return type:

None

unregister_observer(callback)

Remove a previously registered GUI callback.

Implements SimulationControl protocol.

If the callback was not registered, this is a no-op (no error raised).

Parameters:

callback (Callable[[int, SimulationSnapshot], None]) – The callback function to remove.

Return type:

None

step(n=1)

Advance simulation by n ticks.

Implements SimulationControl protocol’s step(n) method.

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).

Parameters:

n (int) – Number of ticks to advance. Must be positive. Defaults to 1 for backward compatibility.

Return type:

WorldState

Returns:

The new WorldState after all ticks complete.

Raises:

ValueError – If n <= 0.

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 or zero

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.

get_time_series()

Extract time series records from completed simulation.

Reads accumulated tick dynamics snapshots stored in persistent_context by the step() function at each year boundary. Each snapshot contains county-level economic state computed by TickDynamicsSystem.

Returns:

year, fips, class distribution shares (bourgeoisie_share, petit_bourgeoisie_share, la_share, proletariat_share, lumpen_share), profit_rate, phi_hour, throughput_position, data_source, and Vol I/II/III fields: capital_stock, median_wage, employment (Vol I); circuit_money, circuit_productive, circuit_commodity, liquidity_ratio, realization_crisis (Vol II); surplus_total, interest_payments, ground_rent, profit_of_enterprise, financialization_share, overaccumulation, profit_squeeze (Vol III).

Return type:

List of dicts with keys

Example

>>> sim = Simulation.from_sqlite(["26163"], year=2022, years=[2022])
>>> sim.run(52)
>>> ts = sim.get_time_series()
>>> for record in ts:
...     print(f"{record['year']} {record['fips']}: LA={record['la_share']:.2f}")

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

get_outcome()

Return current game outcome from EndgameDetector if present.

Searches registered observers for an EndgameDetector and returns its current outcome. If no EndgameDetector is registered, returns IN_PROGRESS.

Return type:

GameOutcome

Returns:

GameOutcome enum value indicating current game state.

Example

>>> from babylon.engine.observers import EndgameDetector
>>> detector = EndgameDetector()
>>> sim = Simulation(state, config, observers=[detector])
>>> sim.get_outcome()
<GameOutcome.IN_PROGRESS: 'in_progress'>

get_current_tick()

Return the current tick number.

Implements SimulationState protocol.

Return type:

int

Returns:

Non-negative integer representing the current simulation tick. Tick 0 is the initial state before any step() calls.

get_snapshot()

Return a complete snapshot of the current simulation state.

Implements SimulationState protocol.

The snapshot is immutable - modifying the returned object does not affect the simulation. The tensor_registry reference allows cached tensor data access without database queries.

Return type:

SimulationSnapshot

Returns:

SimulationSnapshot containing all state at the current tick.

get_territory_state(territory_id)

Return the state of a specific territory.

Implements SimulationState protocol.

Parameters:

territory_id (str) – Unique identifier for the territory (FIPS code for counties).

Return type:

TerritoryState | None

Returns:

TerritoryState if the territory exists, None otherwise.

get_hexes_for_territory(territory_id)

Return the H3 indices claimed by a territory.

Implements SimulationState protocol.

Parameters:

territory_id (str) – Unique identifier for the territory.

Return type:

set[str]

Returns:

Set of H3 index strings. Empty set if territory not found.

get_node_by_spatial_index(h3_index)

Return the territory that claims a specific H3 hex (T027).

Implements SimulationState protocol.

This method bridges the spatial representation (H3 hexes used by map visualization like pydeck) to the simulation’s territory model.

Parameters:

h3_index (str) – H3 cell index (15-character lowercase hex string).

Return type:

TerritoryState | None

Returns:

TerritoryState if a territory claims this hex, None otherwise.

Raises:

ValueError – If h3_index is not a valid H3 cell index.

reset()

Reset simulation to initial state (tick 0).

Implements SimulationControl protocol.

Restores the simulation to its state immediately after initialization: - tick = 0 - All territory states reset to initial values - profit_rate returns to initial computed values - WorldState reset to initial state - History cleared

Implementation note: reset() restores CACHED initial state.

Return type:

None

run_until_endgame(max_ticks=1000)

Run simulation until an endgame condition is met or max_ticks reached.

This method runs the simulation step by step, checking after each tick whether the EndgameDetector has detected a game ending condition. It terminates early if an endgame is reached.

Parameters:

max_ticks (int) – Maximum number of ticks to run before returning. Defaults to 1000 to prevent infinite loops.

Returns:

• final_state: The WorldState when simulation stopped

• outcome: GameOutcome indicating why simulation stopped (may be IN_PROGRESS if max_ticks reached without endgame)

Return type:

Tuple of (final_state, outcome)

Raises:

ValueError – If max_ticks is negative.

Example

>>> from babylon.engine.observers import EndgameDetector
>>> detector = EndgameDetector()
>>> sim = Simulation(state, config, observers=[detector])
>>> final_state, outcome = sim.run_until_endgame(max_ticks=100)
>>> if outcome == GameOutcome.REVOLUTIONARY_VICTORY:
...     print("The workers have won!")