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, *, auditor=None)

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

• auditor (Any)

__init__(systems, *, auditor=None)

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.

• auditor (Any) – Optional Spec 062 ConservationAuditor. When provided, runs at end-of-tick (after all systems) and emits any alarm-severity audit rows onto the event bus per FR-047. When None, audit step is skipped — preserves backward compatibility with pre-spec-062 callers.

Return type:

None

property systems: list[System]

Read-only access to registered systems.

property per_system_ms: dict[str, float]

Cumulative wallclock (ms) per system class name.

Spec-065 T074. Reset via reset_per_system_ms() between runs if the same engine instance is reused.

reset_per_system_ms()

Clear the per-system wallclock accumulator (T074).

Return type:

None

property auditor: Any

The conservation auditor, or None if not configured.

run_tick(graph, services, context)

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.

Spec 062, T068: when self._auditor is set, the ConservationAuditor runs after every system completes. Alarm-severity rows are published as ConservationAlarmEvent instances onto the event bus.

Parameters:

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

• services (ServicesProtocol) – ServicesProtocol with config, formulas, event_bus, database, metrics

• context (Union[dict[str, Any], TickContext]) – TickContext or dict passed to all systems

Return type:

None

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)

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:

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 package — Spec 059 US1 / ADR-005 Part B.

Replaces the historical 1048-LOC engine/simulation.py single file with a package whose __init__.py re-exports the Simulation class and its public surface unchanged. The original implementation lives at _legacy.py while the content split into focused sub-components (orchestrator.py / observer_dispatch.py / lifecycle.py / error_recovery.py per data-model.md §2.2) is deferred to a follow-up commit — preserving byte-equality and import equivalence trumps SC-002’s per-file LOC budget for this commit.

Import equivalence (FR-003 / contracts/import-equivalence.md C2): every existing from babylon.engine import Simulation (via the engine package’s PEP 562 __getattr__) and from babylon.engine.simulation import … resolves unchanged via this re-export.

Public API (contracts/protocol-satisfaction.md P2): the facade preserves the 18 public methods captured at pre-Bundle-2 baseline in reports/simulation-public-methods-before.txt.

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.intelligence.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[dict[str, Any]]

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[WorldState, GameOutcome]

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!")

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. Spec 008: Added metrics field for dependency-injected telemetry.

class babylon.engine.services.EconomicsFallbackTally(national_params_observations=0, melt_calculator_wired=False, basket_calculator_wired=False, gamma_calculator_wired=False, melt_unavailable=0, gamma_basket_calculator_none=0, gamma_iii_calculator_none=0, gamma_iii_returned_none=0)

Bases: object

Loud observability for economics-calculator fallbacks (C.8 / spec 2.R).

TickDynamicsSystem substitutes a hardcoded coefficient whenever an economics calculator is unwired (None) or returns no data. Historically these substitutions were silent — a fully-unwired run reported gamma_III = 0.33 forever with no trace of why. This tally records each fallback and the wired-vs-None status of every calculator so the run manifest can attest whether gamma was genuinely computed or merely defaulted.

Pure instrumentation: recording a fallback NEVER changes a computed value. The caller selects the fallback constant first, then calls record_*.

A fresh tally is created per ServiceContainer (default_factory), so counters are scoped to a single run and safe across processes/tests.

Parameters:

• national_params_observations (int)

• melt_calculator_wired (bool)

• basket_calculator_wired (bool)

• gamma_calculator_wired (bool)

• melt_unavailable (int)

• gamma_basket_calculator_none (int)

• gamma_iii_calculator_none (int)

• gamma_iii_returned_none (int)

national_params_observations: int = 0

How many times _compute_national_params ran (year boundaries seen).

melt_calculator_wired: bool = False

Wired-vs-None status of each calculator (snapshot; last observation wins).

basket_calculator_wired: bool = False

gamma_calculator_wired: bool = False

melt_unavailable: int = 0

Per-fallback counters.

gamma_basket_calculator_none: int = 0

gamma_iii_calculator_none: int = 0

gamma_iii_returned_none: int = 0

observe_wiring(*, melt, basket, gamma)

Record calculator wired-vs-None status for this observation.

Parameters:

• melt (bool) – Whether melt_calculator is wired (not None).

• basket (bool) – Whether basket_calculator is wired (not None).

• gamma (bool) – Whether gamma_calculator is wired (not None).

Return type:

None

record_melt_unavailable()

Count a MELT-unavailable early return (calculator wired but no data).

Return type:

None

record_gamma_basket_calculator_none()

Count a gamma_basket fallback taken because the calculator is None.

Return type:

None

record_gamma_iii_calculator_none()

Count a gamma_III fallback taken because the calculator is None.

Return type:

None

record_gamma_iii_returned_none()

Count a gamma_III fallback taken because a wired calculator returned no data.

Return type:

None

to_dict()

Serialize to a manifest-ready dict (stable key order).

Return type:

dict[str, int | bool]

Returns:

Dict of counter/status fields for the manifest economics_fallbacks block.

__init__(national_params_observations=0, melt_calculator_wired=False, basket_calculator_wired=False, gamma_calculator_wired=False, melt_unavailable=0, gamma_basket_calculator_none=0, gamma_iii_calculator_none=0, gamma_iii_returned_none=0)

Parameters:

• national_params_observations (int)

• melt_calculator_wired (bool)

• basket_calculator_wired (bool)

• gamma_calculator_wired (bool)

• melt_unavailable (int)

• gamma_basket_calculator_none (int)

• gamma_iii_calculator_none (int)

• gamma_iii_returned_none (int)

Return type:

None

class babylon.engine.services.ServiceContainer(config, database, event_bus, formulas, defines, metrics, field_registry=None, opposition_registry=None, reserve_army_data_source=None, dispossession_data_source=None, productivity_data_source=None, melt_calculator=None, basket_calculator=None, gamma_calculator=None, capital_calculator=None, throughput_calculator=None, transition_engine=None, tensor_registry=None, economics_fallbacks=<factory>, community_hypergraph=None, turnover_profile_source=None, inventory_data_source=None, depreciation_data_source=None, hex_grid=None, persistence=None, tracer=None, boundary_register=None, auditor=None, distribution_calculator=None, interest_calculator=None, credit_cycle_detector=None, fictitious_capital_calculator=None, rent_calculator=None, housing_calculator=None, counter_tendency_calculator=None, value_basis_converter=None, financial_crisis_assessor=None, z1_source=None, housing_data_source=None, periphery_labor_source=None, final_demand_source=None, industry_county_allocator=None, production_chain_calculator=None, bea_industries=None)

Bases: object

Container for all simulation services.

Aggregates the six core services needed by the simulation, plus optional economics calculator services for tick dynamics (Feature 017):

Core:

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

• metrics: Telemetry collector for observability (Spec 008)

Field Topology (Feature 002, optional for backward compatibility):

• field_registry: Contradiction field computation registry

Economics (Feature 017, all optional for backward compatibility):

• melt_calculator: National MELT computation (Feature 013)

• basket_calculator: Basket visibility computation (Feature 013)

• gamma_calculator: Reproductive visibility computation (Feature 015)

• capital_calculator: Capital stock computation (Feature 012)

• throughput_calculator: Throughput position computation (Feature 014)

• transition_engine: Class transition engine (Feature 016)

• tensor_registry: Cached economic tensor data (Feature 011)

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
>>> container.metrics.increment("ticks_processed")

Parameters:

• config (SimulationConfig)

• database (DatabaseProtocol)

• event_bus (EventBus)

• formulas (FormulaRegistry)

• defines (GameDefines)

• metrics (MetricsCollectorProtocol)

• field_registry (Any)

• opposition_registry (Any)

• reserve_army_data_source (Any)

• dispossession_data_source (Any)

• productivity_data_source (Any)

• melt_calculator (Any)

• basket_calculator (Any)

• gamma_calculator (Any)

• capital_calculator (Any)

• throughput_calculator (Any)

• transition_engine (Any)

• tensor_registry (Any)

• economics_fallbacks (EconomicsFallbackTally)

• community_hypergraph (Any)

• turnover_profile_source (Any)

• inventory_data_source (Any)

• depreciation_data_source (Any)

• hex_grid (Any)

• persistence (Any)

• tracer (Any)

• boundary_register (Any)

• auditor (Any)

• distribution_calculator (Any)

• interest_calculator (Any)

• credit_cycle_detector (Any)

• fictitious_capital_calculator (Any)

• rent_calculator (Any)

• housing_calculator (Any)

• counter_tendency_calculator (Any)

• value_basis_converter (Any)

• financial_crisis_assessor (Any)

• z1_source (Any)

• housing_data_source (Any)

• periphery_labor_source (Any)

• final_demand_source (Any)

• industry_county_allocator (Any)

• production_chain_calculator (Any)

• bea_industries (list[str] | None)

config: SimulationConfig

database: DatabaseProtocol

event_bus: EventBus

formulas: FormulaRegistry

defines: GameDefines

metrics: MetricsCollectorProtocol

field_registry: Any = None

opposition_registry: Any = None

reserve_army_data_source: Any = None

dispossession_data_source: Any = None

productivity_data_source: Any = None

melt_calculator: Any = None

basket_calculator: Any = None

gamma_calculator: Any = None

capital_calculator: Any = None

throughput_calculator: Any = None

transition_engine: Any = None

tensor_registry: Any = None

economics_fallbacks: EconomicsFallbackTally

community_hypergraph: Any = None

turnover_profile_source: Any = None

inventory_data_source: Any = None

depreciation_data_source: Any = None

hex_grid: Any = None

persistence: Any = None

tracer: Any = None

boundary_register: Any = None

auditor: Any = None

distribution_calculator: Any = None

interest_calculator: Any = None

credit_cycle_detector: Any = None

fictitious_capital_calculator: Any = None

rent_calculator: Any = None

housing_calculator: Any = None

counter_tendency_calculator: Any = None

value_basis_converter: Any = None

financial_crisis_assessor: Any = None

z1_source: Any = None

housing_data_source: Any = None

periphery_labor_source: Any = None

final_demand_source: Any = None

industry_county_allocator: Any = None

production_chain_calculator: Any = None

bea_industries: list[str] | None = None

The configured BEA Summary industry list — defines the alignment baseline for FR-006 (industry-list mismatch fail-fast). Set at scenario-load time; None until then (the Spec 057 pipeline falls back to graceful-degradation stub behavior when None per data-model.md ServiceContainer notes).

__init__(config, database, event_bus, formulas, defines, metrics, field_registry=None, opposition_registry=None, reserve_army_data_source=None, dispossession_data_source=None, productivity_data_source=None, melt_calculator=None, basket_calculator=None, gamma_calculator=None, capital_calculator=None, throughput_calculator=None, transition_engine=None, tensor_registry=None, economics_fallbacks=<factory>, community_hypergraph=None, turnover_profile_source=None, inventory_data_source=None, depreciation_data_source=None, hex_grid=None, persistence=None, tracer=None, boundary_register=None, auditor=None, distribution_calculator=None, interest_calculator=None, credit_cycle_detector=None, fictitious_capital_calculator=None, rent_calculator=None, housing_calculator=None, counter_tendency_calculator=None, value_basis_converter=None, financial_crisis_assessor=None, z1_source=None, housing_data_source=None, periphery_labor_source=None, final_demand_source=None, industry_county_allocator=None, production_chain_calculator=None, bea_industries=None)

Parameters:

• config (SimulationConfig)

• database (DatabaseProtocol)

• event_bus (EventBus)

• formulas (FormulaRegistry)

• defines (GameDefines)

• metrics (MetricsCollectorProtocol)

• field_registry (Any)

• opposition_registry (Any)

• reserve_army_data_source (Any)

• dispossession_data_source (Any)

• productivity_data_source (Any)

• melt_calculator (Any)

• basket_calculator (Any)

• gamma_calculator (Any)

• capital_calculator (Any)

• throughput_calculator (Any)

• transition_engine (Any)

• tensor_registry (Any)

• economics_fallbacks (EconomicsFallbackTally)

• community_hypergraph (Any)

• turnover_profile_source (Any)

• inventory_data_source (Any)

• depreciation_data_source (Any)

• hex_grid (Any)

• persistence (Any)

• tracer (Any)

• boundary_register (Any)

• auditor (Any)

• distribution_calculator (Any)

• interest_calculator (Any)

• credit_cycle_detector (Any)

• fictitious_capital_calculator (Any)

• rent_calculator (Any)

• housing_calculator (Any)

• counter_tendency_calculator (Any)

• value_basis_converter (Any)

• financial_crisis_assessor (Any)

• z1_source (Any)

• housing_data_source (Any)

• periphery_labor_source (Any)

• final_demand_source (Any)

• industry_county_allocator (Any)

• production_chain_calculator (Any)

• bea_industries (list[str] | None)

Return type:

None

classmethod create(config=None, defines=None, metrics=None, *, hex_grid=None, persistence=None, tracer=None, reserve_army_data_source=None, dispossession_data_source=None, productivity_data_source=None, field_registry=None, opposition_registry=None, melt_calculator=None, basket_calculator=None, gamma_calculator=None, capital_calculator=None, throughput_calculator=None, transition_engine=None, tensor_registry=None, community_hypergraph=None, turnover_profile_source=None, inventory_data_source=None, depreciation_data_source=None, distribution_calculator=None, interest_calculator=None, credit_cycle_detector=None, fictitious_capital_calculator=None, rent_calculator=None, housing_calculator=None, counter_tendency_calculator=None, value_basis_converter=None, financial_crisis_assessor=None, z1_source=None, housing_data_source=None, periphery_labor_source=None, final_demand_source=None, industry_county_allocator=None, production_chain_calculator=None, bea_industries=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.

• metrics (MetricsCollectorProtocol | None) – Optional custom metrics collector. If None, creates a new MetricsCollector instance. Pass a mock for testing.

• field_registry (Any) – Optional FieldRegistry for contradiction fields (Feature 002).

• melt_calculator (Any) – Optional MELTCalculator (Feature 013).

• basket_calculator (Any) – Optional BasketVisibilityCalculator (Feature 013).

• gamma_calculator (Any) – Optional GammaIIICalculator (Feature 015).

• capital_calculator (Any) – Optional CapitalStockCalculator (Feature 012).

• throughput_calculator (Any) – Optional ThroughputCalculator (Feature 014).

• transition_engine (Any) – Optional ClassTransitionEngine (Feature 016).

• tensor_registry (Any) – Optional TensorRegistry for cached tensor data (Feature 011).

• community_hypergraph (Any) – Optional XGI Hypergraph for community membership (Feature 022).

• hex_grid (Any)

• persistence (Any)

• tracer (Any)

• reserve_army_data_source (Any)

• dispossession_data_source (Any)

• productivity_data_source (Any)

• opposition_registry (Any)

• turnover_profile_source (Any)

• inventory_data_source (Any)

• depreciation_data_source (Any)

• distribution_calculator (Any)

• interest_calculator (Any)

• credit_cycle_detector (Any)

• fictitious_capital_calculator (Any)

• rent_calculator (Any)

• housing_calculator (Any)

• counter_tendency_calculator (Any)

• value_basis_converter (Any)

• financial_crisis_assessor (Any)

• z1_source (Any)

• housing_data_source (Any)

• periphery_labor_source (Any)

• final_demand_source (Any)

• industry_county_allocator (Any)

• production_chain_calculator (Any)

• bea_industries (list[str] | None)

Return type:

ServiceContainer

Returns:

ServiceContainer with all services initialized

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.kernel.event_bus.Event(type, tick, payload, timestamp=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))

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

Deterministic sim-time derived from tick (Constitution III.7)

type: str

tick: int

payload: dict[str, Any]

timestamp: datetime = datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)

__post_init__()

Derive the default timestamp from tick (Constitution III.7).

Return type:

None

__init__(type, tick, payload, timestamp=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))

Parameters:

• type (str)

• tick (int)

• payload (dict[str, Any])

• timestamp (datetime)

Return type:

None

class babylon.kernel.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.kernel.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()
>>> la = registry.get("labor_aristocracy_ratio")
>>> result = la(core_wages=120.0, value_produced=100.0)

__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 formulas from babylon.formulas: - 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.

Spec-066 baseline (placeholder) per ADR043 + data-model.md section 2: the bridged runner passes IdeologicalProfile(class_consciousness=0.1, national_identity=0.5) to every entity. The bridge’s ternary mapping (r = cc * (1 - ni), f = ni * (1 - cc), l = 1 - r - f) yields the target placeholder (r=0.05, l=0.50, f=0.45) per Clarifications Q3. The rejected high-cc / high-ni alternative (cc=0.5, ni=0.9) is theoretically dubious: Marx treats class consciousness and national identity as antagonistic, so co-existing high values are unstable. Per-county data-driven seeding is deferred to a future spec; the placeholder must remain explicit and uniform across all 83 x 2 = 166 entities until that work lands.

babylon.engine.factories.create_proletariat(id=PERIPHERY_WORKER_ID, name='Proletariat', wealth=0.5, ideology=None, 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, county_fips=None)

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 | None) – Ideological position. Accepts float (legacy, scalar -1=revolutionary..+1=reactionary), IdeologicalProfile (spec-066 placeholder + future per-county data), or None to use the legacy default -0.3. The spec-066 bridged runner passes IdeologicalProfile(class_consciousness=0.1, national_identity=0.5) to every county entity to materialize the placeholder (r=0.05, l=0.50, f=0.45).

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

• county_fips (str | None)

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_labor_aristocracy(id=LABOR_ARISTOCRACY_ID, name='Labor Aristocracy', wealth=0.8, ideology=None, organization=0.05, repression_faced=0.3, subsistence_threshold=0.3, p_acquiescence=0.0, p_revolution=0.0, description='Super-waged core working class (pacified by imperial rent)', effective_wealth=0.0, unearned_increment=0.0, ppp_multiplier=1.0, county_fips=None)

Create a labor-aristocracy (super-waged core worker) social class.

Per Cope (Divided World Divided Class), Amin (The Law of Worldwide Value), and the project’s Fundamental Theorem (W_c > V_c): the core working class receives wages above the value it produces, funded by imperial rent — pacifying it (P(S|A) > P(S|R)) until the rent pool decays. Defaults mirror create_labor_aristocracy_scenario in the legacy scenarios (well-off worker, very low organization, low repression needed).

The LA is defined by:

• LABOR_ARISTOCRACY role (employed producer in the Amin/Wallerstein circuit: production routes to the employer via the incoming WAGES edge; the wages phase pays back productivity + super-wage bonus)

• Wealth above subsistence (default 0.8 vs threshold 0.3)

• Very low organization (0.05) — the bribe substitutes for solidarity

• Low repression faced (0.3) — hegemony, not force

Parameters:

• id (str) – Unique identifier matching ^C[0-9]{3}$ pattern.

• name (str) – Human-readable name.

• wealth (float) – Economic resources (default 0.8 — above subsistence).

• ideology (float | IdeologicalProfile | None) – Ideological position (float, IdeologicalProfile, or None for the legacy scalar default, as in create_proletariat()).

• organization (float) – Collective cohesion (default 0.05).

• repression_faced (float) – State violence level (default 0.3).

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

• p_acquiescence (float) – P(S|A) (default 0.0, calculated by engine).

• p_revolution (float) – P(S|R) (default 0.0, calculated by engine).

• description (str) – Optional description.

• effective_wealth (float) – PPP-adjusted wealth (calculated by engine).

• unearned_increment (float) – PPP bonus (calculated by engine).

• ppp_multiplier (float) – PPP multiplier applied to wages.

• county_fips (str | None) – Optional 5-digit county attribution (bridged runs).

Return type:

SocialClass

Returns:

SocialClass configured as labor aristocracy.

Example

>>> worker = create_labor_aristocracy()
>>> worker.role
<SocialRole.LABOR_ARISTOCRACY: 'labor_aristocracy'>
>>> worker.wealth > worker.subsistence_threshold
True

babylon.engine.factories.create_bourgeoisie(id=COMPRADOR_ID, name='Bourgeoisie', wealth=10.0, ideology=None, 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, county_fips=None)

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 | None) – Ideological position. Accepts float (legacy, scalar -1=revolutionary..+1=reactionary), IdeologicalProfile (spec-066 placeholder + future per-county data), or None to use the legacy default 0.8. The spec-066 bridged runner passes IdeologicalProfile(class_consciousness=0.1, national_identity=0.5) to every county entity to materialize the placeholder (r=0.05, l=0.50, f=0.45).

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

• county_fips (str | None)

Return type:

SocialClass

Returns:

SocialClass configured as bourgeoisie

Example

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

babylon.engine.factories.create_contradiction_frame(scope='global')

Create a ContradictionFrame for the given simulation scope.

Parameters:

scope (str) – The scope identifier (e.g., “global”, “national”).

Return type:

ContradictionFrame

Returns:

A new ContradictionFrame for the given scope.