babylon.engine

Simulation engine for the Babylon game loop.

This package contains the core game loop logic: - simulation_engine: The step() function for state transformation - simulation: Simulation facade class for multi-tick runs with history - scenarios: Factory functions for creating initial states - factories: Entity factory functions (create_proletariat, create_bourgeoisie) - history_formatter: Narrative generation from simulation history - Dependency injection: ServiceContainer, EventBus, FormulaRegistry

Phase 2.1: Refactored to modular System architecture. Sprint 3: Central Committee (Dependency Injection) Sprint 9: Integration proof with Simulation facade

babylon.engine.step(state, config, persistent_context=None, defines=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.

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)

class babylon.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 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)[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

run_tick(graph, services, context)[source]

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

Return type:

None

property systems: list[System]: Read-only access to registered systems.

class babylon.engine.Simulation(initial_state, config, observers=None, defines=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)

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

Return type:

None

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

property config: SimulationConfig: Return the simulation configuration.

property current_state: WorldState: Return the current WorldState.

property defines: GameDefines: Return the game defines.

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

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.

remove_observer(observer)[source]

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

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

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

property services: ServiceContainer: Return the persistent ServiceContainer.

step()[source]

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.

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.

class babylon.engine.AsyncSimulationRunner(simulation, tick_interval=1.0)[source]

Bases: object

Async runner that decouples UI from simulation engine.

The runner wraps a Simulation instance and provides:

Non-blocking steps: Uses asyncio.to_thread() to run simulation.step() without blocking the event loop.
State queue: Pushes WorldState snapshots to an async queue for the UI to consume at its own pace.
Continuous play: start()/stop() manage a background loop that steps the simulation at configurable intervals.
Queue overflow handling: Drops oldest states when queue is full (MAX_QUEUE_SIZE=10) to prevent memory issues.

Parameters:

simulation (Simulation)
tick_interval (float)

MAX_QUEUE_SIZE: Maximum states to buffer before dropping oldest.

Example

>>> runner = AsyncSimulationRunner(sim, tick_interval=0.5)
>>> state = await runner.step_once()  # Single step
>>> print(f"Now at tick {state.tick}")

Thread Safety:: The runner uses an asyncio.Lock to serialize step_once() calls, ensuring simulation state consistency even with concurrent access.

MAX_QUEUE_SIZE: int = 10

__init__(simulation, tick_interval=1.0)[source]

Initialize the async runner with a simulation.

Parameters:

simulation (Simulation) – The Simulation facade to run.
tick_interval (float) – Seconds between steps in continuous mode. Must be positive.

Raises:

ValueError – If tick_interval is not positive.

Return type:

None

async drain_queue()[source]

Remove and return all states from the queue.

Return type:: list[WorldState]
Returns:: List of all WorldState objects that were in the queue, in order from oldest to newest. Empty list if queue was empty.

async get_state()[source]

Get a state from the queue without blocking.

Return type:: WorldState | None
Returns:: The next WorldState if available, None otherwise.

async get_state_blocking(timeout=None)[source]

Get a state from the queue, waiting if necessary.

Parameters:: timeout (float | None) – Maximum seconds to wait. None means wait forever.
Return type:: WorldState
Returns:: The next WorldState from the queue.
Raises:: asyncio.TimeoutError – If timeout expires before a state is available.

property is_running: bool: Return True if continuous mode is active.

property queue: Queue[WorldState]

Return the state queue for direct access.

Returns:: The asyncio.Queue containing WorldState snapshots.

async reset(new_simulation)[source]

Reset the runner with a new simulation.

Stops the runner if running, drains the queue, and assigns the new simulation for future steps.

Parameters:: new_simulation (Simulation) – The new Simulation instance to use.
Return type:: None

property simulation: Simulation: Return the wrapped Simulation instance.

async start()[source]

Start continuous simulation mode.

Creates a background task that steps the simulation at tick_interval intervals. Idempotent - calling start() when already running is a no-op.

The task reference is stored to prevent garbage collection.

Return type:: None

async step_once()[source]

Execute a single simulation step without blocking.

Uses asyncio.to_thread() to run simulation.step() in a thread pool, keeping the event loop responsive.

The new state is pushed to the queue. If the queue is full, the oldest state is dropped to make room.

Return type:: WorldState
Returns:: The new WorldState after the step.

Note

Uses an asyncio.Lock to serialize concurrent calls, ensuring simulation state consistency.

async stop()[source]

Stop continuous simulation mode.

Cancels the background task and waits for it to finish. Idempotent - calling stop() when not running is a no-op.

Return type:: None

property tick_interval: float: Return the interval between steps in continuous mode.

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

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

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

babylon.engine.format_class_struggle_history(simulation)[source]

Format simulation history as a narrative of class struggle.

Generates a human-readable summary of the simulation history, highlighting wealth transfers, ideological changes, and tension accumulation.

Parameters:: simulation (Simulation) – A Simulation instance with history data
Return type:: str
Returns:: A formatted string narrative describing the class struggle dynamics.

Example

>>> sim = Simulation(initial_state, config)
>>> sim.run(100)
>>> print(format_class_struggle_history(sim))
# History of Class Struggle
...

babylon.engine.create_two_node_scenario(worker_wealth=0.5, owner_wealth=0.5, extraction_efficiency=0.8, repression_level=0.5, worker_organization=0.1, worker_ideology=0.0)[source]

Create the minimal viable dialectic: one worker, one owner, one exploitation edge.

This is the two-node scenario from the Phase 1 blueprint, now ready for Phase 2 simulation. It models the fundamental class relationship: - Worker produces value (source of exploitation edge) - Owner extracts imperial rent (target of exploitation edge) - Tension accumulates on the edge

Parameters:

worker_wealth (float) – Initial wealth for periphery worker (default 0.5)
owner_wealth (float) – Initial wealth for core owner (default 0.5)
extraction_efficiency (float) – Alpha in imperial rent formula (default 0.8)
repression_level (float) – State violence capacity (default 0.5)
worker_organization (float) – Worker class cohesion (default 0.1)
worker_ideology (float) – Worker ideology, -1=revolutionary to +1=reactionary (default 0.0)

Return type:

tuple[WorldState, SimulationConfig, GameDefines]

Returns:

Tuple of (WorldState, SimulationConfig, GameDefines) ready for step() function.

Example

>>> state, config, defines = create_two_node_scenario()
>>> for _ in range(100):
...     state = step(state, config)
>>> print(f"Worker wealth after 100 ticks: {state.entities['C001'].wealth}")

babylon.engine.create_high_tension_scenario()[source]

Create a scenario with high initial tension.

Worker is poor, owner is rich, tension is already elevated. Useful for testing phase transitions and rupture conditions.

Return type:: tuple[WorldState, SimulationConfig, GameDefines]
Returns:: Tuple of (WorldState, SimulationConfig, GameDefines) near rupture point.

babylon.engine.create_labor_aristocracy_scenario()[source]

Create a scenario with a labor aristocracy (Wc > Vc).

Worker receives more than they produce, enabled by imperial rent from elsewhere. Tests consciousness decay mechanics.

Return type:: tuple[WorldState, SimulationConfig, GameDefines]
Returns:: Tuple of (WorldState, SimulationConfig, GameDefines) with labor aristocracy.

class babylon.engine.Event(type, tick, payload, timestamp=<factory>)[source]

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

__init__(type, tick, payload, timestamp=<factory>)

Parameters:

type (str)
tick (int)
payload (dict[str, Any])
timestamp (datetime)

Return type:

None

type: str

tick: int

payload: dict[str, Any]

timestamp: datetime

class babylon.engine.EventBus[source]

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

Initialize an empty event bus.

Return type:: None

clear_blocked_events()[source]

Remove all blocked event records.

Return type:: None

clear_history()[source]

Remove all events from history.

Return type:: None

get_blocked_events()[source]

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.

get_history()[source]

Get a copy of all published events.

Return type:: list[Event]
Returns:: List of events in chronological order (oldest first).

property interceptor_count: int: Number of registered interceptors.

publish(event, context=None)[source]

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

register_interceptor(interceptor)[source]

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)

subscribe(event_type, handler)[source]

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

unregister_interceptor(interceptor)[source]

Remove an interceptor from the chain.

Parameters:: interceptor (EventInterceptor) – The interceptor to remove.
Raises:: ValueError – If the interceptor is not registered.
Return type:: None

class babylon.engine.DatabaseConnection(url='sqlite:///babylon.db')[source]

Bases: object

Injectable database connection wrapper.

Wraps SQLAlchemy engine and sessionmaker to provide clean resource management and testability.

Example

>>> db = DatabaseConnection(url="sqlite:///:memory:")
>>> with db.session() as session:
...     result = session.execute(text("SELECT 1"))
...     print(result.scalar())
1
>>> db.close()

Parameters:: url (str)

__init__(url='sqlite:///babylon.db')[source]

Initialize database connection.

Parameters:: url (str) – SQLAlchemy database URL. Defaults to local SQLite file. Use “sqlite:///:memory:” for in-memory testing.
Return type:: None

close()[source]

Dispose of the database engine and release resources.

After calling close(), the connection cannot be used. Attempting to create new sessions will fail.

Return type:: None

session()[source]

Get a database session within a context manager.

The session is automatically closed when the context exits. On exception, the session is rolled back.

Yields:: SQLAlchemy Session object
Return type:: Iterator[Session]

Example

>>> with db.session() as session:
...     session.execute(text("INSERT INTO ..."))
...     session.commit()

Return type:: Iterator[Session]

class babylon.engine.FormulaRegistry[source]

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

Initialize an empty formula registry.

Return type:: None

classmethod default()[source]

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

get(name)[source]

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

List all registered formula names.

Return type:: list[str]
Returns:: List of formula names in arbitrary order

register(name, func)[source]

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

class babylon.engine.ServiceContainer(config, database, event_bus, formulas, defines)[source]

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)

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

Parameters:

config (SimulationConfig)
database (DatabaseConnection)
event_bus (EventBus)
formulas (FormulaRegistry)
defines (GameDefines)

Return type:

None

classmethod create(config=None, defines=None)[source]

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

config: SimulationConfig

database: DatabaseConnection

event_bus: EventBus

formulas: FormulaRegistry

defines: GameDefines

class babylon.engine.SimulationObserver(*args, **kwargs)[source]

Bases: Protocol

Protocol for entities observing simulation state changes.

Observers receive notifications at three lifecycle points: 1. on_simulation_start() - when simulation begins (first step) 2. on_tick() - after each tick completes 3. on_simulation_end() - when simulation ends (explicit end() call)

Note: All state objects (WorldState) are frozen and immutable. Attempting to modify them will raise AttributeError.

Example

>>> class MyObserver:
...     @property
...     def name(self) -> str:
...         return "MyObserver"
...
...     def on_simulation_start(
...         self, initial_state: WorldState, config: SimulationConfig
...     ) -> None:
...         print(f"Started at tick {initial_state.tick}")
...
...     def on_tick(
...         self, previous_state: WorldState, new_state: WorldState
...     ) -> None:
...         print(f"Tick {previous_state.tick} -> {new_state.tick}")
...
...     def on_simulation_end(self, final_state: WorldState) -> None:
...         print(f"Ended at tick {final_state.tick}")

__init__(*args, **kwargs)

property name: str

Observer identifier for logging and debugging.

Returns:: A string identifying this observer instance.

on_simulation_end(final_state)[source]

Called when simulation ends (on explicit end() call).

Use this hook to cleanup resources, generate summaries, or finalize any accumulated data.

Parameters:: final_state (WorldState) – The final WorldState when simulation ends.
Return type:: None

on_simulation_start(initial_state, config)[source]

Called when simulation begins (on first step call).

Use this hook to initialize resources, establish context, or prepare for observing the simulation run.

Parameters:

initial_state (WorldState) – The WorldState at tick 0 (before any steps).
config (SimulationConfig) – The SimulationConfig for this run.

Return type:

None

on_tick(previous_state, new_state)[source]

Called after each tick completes with both states for delta analysis.

This is the primary notification hook. Observers receive both the previous and new state to enable delta analysis (what changed).

Parameters:

previous_state (WorldState) – WorldState before the tick.
new_state (WorldState) – WorldState after the tick.

Return type:

None

class babylon.engine.TopologyMonitor(resilience_test_interval=5, resilience_removal_rate=DEFAULT_REMOVAL_RATE, logger=None, gaseous_threshold=None, condensation_threshold=None, vanguard_threshold=None)[source]

Bases: object

Observer tracking solidarity network condensation.

Implements SimulationObserver protocol to receive state change notifications and analyze the topology of SOLIDARITY edges.

Monitors:

Connected components (atomization vs. condensation)
Percolation ratio (L_max / N)
Liquidity metrics (potential vs. actual solidarity)
Resilience (survives targeted node removal)

Narrative states logged:

Gaseous: percolation < 0.1 (atomized)
Liquid: percolation crosses 0.5 (condensation detected)
Brittle: potential >> actual (broad but fragile)
Fragile: resilience = False (Sword of Damocles)

Parameters:

resilience_test_interval (int)
resilience_removal_rate (float)
logger (logging.Logger | None)
gaseous_threshold (float | None)
condensation_threshold (float | None)
vanguard_threshold (float | None)

name: Observer identifier (“TopologyMonitor”)

history: List of TopologySnapshot for each tick

__init__(resilience_test_interval=5, resilience_removal_rate=DEFAULT_REMOVAL_RATE, logger=None, gaseous_threshold=None, condensation_threshold=None, vanguard_threshold=None)[source]

Initialize TopologyMonitor.

Parameters:

resilience_test_interval (int) – Run resilience test every N ticks (0 = disabled). Default 5.
resilience_removal_rate (float) – Fraction of nodes to remove in test. Default 0.2 (20%).
logger (Logger | None) – Logger instance (default: module logger)
gaseous_threshold (float | None) – Percolation ratio below this = atomized. Defaults to GameDefines.topology.gaseous_threshold (0.1).
condensation_threshold (float | None) – Percolation ratio for phase transition. Defaults to GameDefines.topology.condensation_threshold (0.5).
vanguard_threshold (float | None) – Cadre density threshold for solid phase. Defaults to GameDefines.topology.vanguard_density_threshold (0.5).

Return type:

None

get_pending_events()[source]

Return and clear pending events for collection by Simulation facade.

Observer events cannot be emitted directly to WorldState because observers run AFTER WorldState is frozen. Instead, pending events are collected by the Simulation facade and injected into the NEXT tick’s WorldState.

Return type:: list[SimulationEvent]
Returns:: List of pending SimulationEvent objects (cleared after return).

property history: list[TopologySnapshot]: Return copy of snapshot history.

property name: str: Return observer identifier.

on_simulation_end(_final_state)[source]

Called when simulation ends.

Logs summary of topology metrics.

Parameters:: _final_state (WorldState) – Final WorldState when simulation ends (unused)
Return type:: None

on_simulation_start(initial_state, _config)[source]

Called when simulation begins.

Initializes history and records initial topology snapshot.

Parameters:

initial_state (WorldState) – WorldState at tick 0
_config (SimulationConfig) – SimulationConfig for this run (unused)

Return type:

None

on_tick(_previous_state, new_state)[source]

Called after each tick completes.

Records topology snapshot and detects phase transitions.

Parameters:

_previous_state (WorldState) – WorldState before the tick (unused)
new_state (WorldState) – WorldState after the tick

Return type:

None

Modules

`context`	Context models for simulation tick execution.
`database`	Injectable database connection for the simulation engine.
`event_bus`	Event system for decoupled communication in the simulation.
`event_evaluator`	Event Template evaluation engine.
`factories`	Factory functions for creating simulation entities.
`formula_registry`	Formula registry for hot-swappable mathematical functions.
`history`	History Stack module for the Babylon simulation engine.
`history_formatter`	History formatter for generating narrative summaries.
`interceptor`	Event Interceptor pattern for Epoch 2 adversarial mechanics.
`observer`	Observer protocol for simulation state change notifications.
`observers`	Observer implementations for simulation state monitoring.
`runner`	Async simulation runner for non-blocking UI integration.
`scenarios`	Factory functions for creating simulation scenarios.
`services`	Service container for dependency injection.
`simulation`	Simulation facade class for running multi-tick simulations.
`simulation_engine`	Simulation engine for the Babylon game loop.
`systems`	Simulation systems for the Babylon engine.
`topology_monitor`	Topology Monitor for phase transition detection (Sprint 3.1, 3.3).