babylon.engine.topology_monitor

Topology Monitor for phase transition detection (Sprint 3.1, 3.3).

The TopologyMonitor is a SimulationObserver that tracks the “condensation” of revolutionary consciousness through the social graph using percolation theory. It detects phase transitions between 4 phases of movement organization.

Theoretical Model (4-Phase):

  • Gaseous State: percolation < 0.1 (atomized, no coordination)

  • Transitional State: 0.1 <= percolation < 0.5 (emerging structure)

  • Liquid State: percolation >= 0.5, cadre_density < 0.5 (mass movement)

  • Solid State: percolation >= 0.5, cadre_density >= 0.5 (vanguard party)

Key Metrics:

  • num_components: Number of disconnected solidarity cells

  • max_component_size (L_max): Largest connected component

  • percolation_ratio: L_max / N (giant component dominance)

  • potential_liquidity: SOLIDARITY edges > 0.1 (sympathizers)

  • actual_liquidity: SOLIDARITY edges > 0.5 (cadre)

  • cadre_density: actual_liquidity / max(1, potential_liquidity)

Narrative States:

  • “Gaseous”: percolation < 0.1 (atomized, vulnerable)

  • “Transitional”: 0.1 <= percolation < 0.5 (emerging structure)

  • “Liquid”: percolation >= 0.5 (mass movement with weak ties)

  • “Solid”: percolation >= 0.5 AND cadre_density >= 0.5 (vanguard party)

  • “Brittle”: potential >> actual (broad but lacks discipline)

  • “Sword of Damocles”: resilience test fails (purge would destroy)

Functions

calculate_component_metrics(...)

Calculate connected component metrics for solidarity network.

calculate_liquidity(G)

Calculate liquidity metrics (potential vs actual solidarity).

check_resilience(G[, removal_rate, ...])

Test if solidarity network survives targeted node removal.

extract_solidarity_subgraph(G[, min_strength])

Extract undirected solidarity network from WorldState graph.

Classes

TopologyMonitor([resilience_test_interval, ...])

Observer tracking solidarity network condensation.
babylon.engine.topology_monitor.extract_solidarity_subgraph(G, min_strength=0.0)[source]

Extract undirected solidarity network from WorldState graph.

Creates an undirected graph containing only social_class nodes and SOLIDARITY edges above the minimum strength threshold. Used for connected component analysis.

Parameters:

  • G – Directed graph from WorldState.to_graph()

  • min_strength – Minimum solidarity_strength to include edge (default 0)

Returns:

Undirected Graph containing only solidarity connections. Isolated social_class nodes are included.

Return type:

nx.Graph[str]

Note

Territory nodes are excluded as they represent spatial substrate, not class positions in the solidarity network.

babylon.engine.topology_monitor.calculate_component_metrics(solidarity_graph, total_social_classes)[source]

Calculate connected component metrics for solidarity network.

Parameters:

  • solidarity_graph – Undirected graph of solidarity connections

  • total_social_classes – Total number of social_class nodes in system

Returns:

Tuple of (num_components, max_component_size, percolation_ratio) - num_components: Number of disconnected subgraphs - max_component_size: Size of largest component (L_max) - percolation_ratio: L_max / N (clamped to [0, 1])

Return type:

tuple[int, int, float]

babylon.engine.topology_monitor.calculate_liquidity(G)[source]

Calculate liquidity metrics (potential vs actual solidarity).

Measures the strength of the solidarity network by counting edges at different thresholds: - Potential (sympathizers): edges > 0.1 strength - Actual (cadre): edges > 0.5 strength

Parameters:

G – Directed graph from WorldState.to_graph()

Returns:

Tuple of (potential_liquidity, actual_liquidity)

Return type:

tuple[int, int]

babylon.engine.topology_monitor.check_resilience(G, removal_rate=DEFAULT_REMOVAL_RATE, survival_threshold=DEFAULT_SURVIVAL_THRESHOLD, seed=None)[source]

Test if solidarity network survives targeted node removal.

Simulates a “purge” by removing a percentage of nodes and checking if the giant component survives. This is the “Sword of Damocles” test - a fragile network can be destroyed by targeting key members.

Parameters:

  • G – Directed graph from WorldState.to_graph()

  • removal_rate – Fraction of nodes to remove (default 0.2 = 20%)

  • survival_threshold – Required fraction of original L_max to survive

  • seed – RNG seed for reproducibility (None = random)

Returns:

ResilienceResult with is_resilient flag and metrics.

Return type:

ResilienceResult

Note

The original graph is NOT modified. Test operates on a copy.

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

property name: str

Return observer identifier.

property history: list[TopologySnapshot]

Return copy of snapshot history.

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

on_simulation_start(initial_state, _config)[source]

Called when simulation begins.

Initializes history and records initial topology snapshot.

Parameters:
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

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