Engine

Core simulation engine and supporting components.

Simulation Engine

Simulation engine for the Babylon game loop.

ADR032: Materialist Causality System Order

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 (materialist causality - base before superstructure): 1. Vitality - Biological cost + death (dead entities don’t work) 2. Territory - Land state updates (land conditions affect production) 3. Production - Value creation (value must exist before extraction) 4. Solidarity - Organization (affects bargaining power) 5. Imperial Rent - Value extraction (landlord eats after harvest) 6. Decomposition - LA decomposes on super-wage crisis (Terminal Crisis) 7. Control Ratio - Guard:prisoner ratio + terminal decision (Terminal Crisis) 8. Metabolism - Environmental degradation (ecological residue of production) 9. Survival - Risk assessment (P(S|A), P(S|R) from material state) 10. Struggle - Action/Revolt (agency responds to survival odds) 11. Consciousness - Ideological drift (ideology responds to material) 12. Contradiction - Tension aggregation (final systemic accounting)

Phase 2.1: Refactored to modular System architecture. Phase 4a: Refactored to use ServiceContainer for dependency injection. ADR032: Reordered systems for materialist causality.

class babylon.engine.simulation_engine.SimulationEngine(systems)[source]

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 materialist causality (ADR032): 1. Vitality (death check) 2. Territory (land state) 3. Production (value creation) 4. Solidarity (organization) 5. Imperial Rent (extraction) 6. Decomposition (LA crisis) 7. Control Ratio (terminal decision) 8. Metabolism (ecology) 9. Survival (risk assessment) 10. Struggle (agency) 11. Consciousness (ideology) 12. Contradiction (tension) 13. ContradictionField (field computation) - Feature 002 14. FieldDerivative (derivatives + principal) - Feature 002 15. EdgeTransition (predicates + state machine) - Feature 002

Parameters:: systems (list[System])

__init__(systems)[source]

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)[source]

Execute all systems in order for one tick.

All logs emitted during this method are automatically tagged with tick number and a unique correlation_id (UUID) for tracing.

Parameters:

graph – NetworkX graph (mutated in place by systems)
services – ServiceContainer with config, formulas, event_bus, database, metrics
context – TickContext or dict passed to all systems

Spec 008: Logs within run_tick() include tick and correlation_id.

babylon.engine.simulation_engine.step(state, config, persistent_context=None, defines=None, calculator_overrides=None)[source]

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.
calculator_overrides (dict[str, Any] | None) – Optional dict of calculator instances to inject into ServiceContainer (e.g., melt_calculator, tensor_registry).

Return type:

WorldState

Returns:

New WorldState at tick + 1

Order encodes historical materialism:

Economic base (value extraction)
Consciousness (responds to material conditions)
Survival calculus (probability updates)
Contradictions (tension from all above)
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

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

class babylon.engine.simulation.Simulation(initial_state, config, observers=None, defines=None, tensor_registry=None, calculator_overrides=None)[source]

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)[source]

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)[source]

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)[source]

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)[source]

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

Parameters:: observer (SimulationObserver) – Observer to remove from notifications.
Return type:: None

register_observer(callback)[source]

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)[source]

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)[source]

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)[source]

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()[source]

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()[source]

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)[source]

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()[source]

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()[source]

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()[source]

Return the current tick number.