Topology System Reference

This reference documents the topology monitoring system that tracks revolutionary organization via percolation theory. For conceptual background, see Percolation Theory & Phase Transitions.

Overview

The topology system analyzes the solidarity subgraph to detect phase transitions in organizational structure. It is implemented as a SimulationObserver that receives state change notifications and computes topological metrics.

Key Components:

babylon.engine.topology_monitor - Observer and helper functions
babylon.models.topology_metrics - Data models for metrics

Constants & Thresholds

Phase Detection

Constant	Value	Purpose
`GASEOUS_THRESHOLD`	0.1	Percolation ratio below this indicates atomized movement
`CONDENSATION_THRESHOLD`	0.5	Crossing this threshold triggers phase shift detection
`BRITTLE_MULTIPLIER`	2	If potential > actual × this, movement is brittle

Liquidity Classification

Constant	Value	Purpose
`POTENTIAL_MIN_STRENGTH`	0.1	Minimum solidarity strength for sympathizer classification
`ACTUAL_MIN_STRENGTH`	0.5	Minimum solidarity strength for cadre classification

Resilience Testing

Constant	Value	Purpose
`DEFAULT_REMOVAL_RATE`	0.2	Fraction of nodes removed in purge simulation (20%)
`DEFAULT_SURVIVAL_THRESHOLD`	0.4	Giant component must survive at 40% of original size

Metrics

TopologySnapshot

The TopologySnapshot model captures metrics at each tick:

Field	Type	Description
`tick`	`int`	Simulation tick number
`num_components`	`int`	Number of disconnected solidarity cells
`max_component_size`	`int`	Size of largest connected component (L_max)
`total_nodes`	`int`	Total social_class nodes in system
`percolation_ratio`	`Probability`	L_max / total_nodes (range [0, 1])
`potential_liquidity`	`int`	Count of SOLIDARITY edges > 0.1 strength
`actual_liquidity`	`int`	Count of SOLIDARITY edges > 0.5 strength
`cadre_density`	`float`	actual/potential liquidity ratio [0, 1] (Sprint 3.3)
`is_resilient`	`bool \| None`	Whether network survives purge test (None if not tested)

ResilienceResult

The ResilienceResult model captures purge simulation output:

Field	Type	Description
`is_resilient`	`bool`	True if network survives simulated purge
`original_max_component`	`int`	L_max before node removal
`post_purge_max_component`	`int`	L_max after removing nodes
`removal_rate`	`float`	Fraction of nodes removed
`survival_threshold`	`float`	Required survival fraction
`seed`	`int \| None`	RNG seed for reproducibility

Functions

extract_solidarity_subgraph

extract_solidarity_subgraph(G, min_strength=0.0)

Extract undirected solidarity network from WorldState graph.

Parameters:

G (nx.DiGraph[str]) – Directed graph from WorldState.to_graph()
min_strength (float) – Minimum solidarity_strength to include edge

Returns:

Undirected graph with only SOLIDARITY edges above threshold

Return type:

nx.Graph[str]

Example:

from babylon.engine.topology_monitor import extract_solidarity_subgraph

graph = state.to_graph()
solidarity_graph = extract_solidarity_subgraph(graph, min_strength=0.1)

calculate_component_metrics

calculate_component_metrics(solidarity_graph, total_social_classes)

Calculate connected component metrics for percolation analysis.

Parameters:

solidarity_graph (nx.Graph[str]) – Undirected solidarity graph
total_social_classes (int) – Total number of social_class nodes

Returns:

Tuple of (num_components, max_component_size, percolation_ratio)

Return type:

tuple[int, int, float]

Example:

from babylon.engine.topology_monitor import (
    extract_solidarity_subgraph,
    calculate_component_metrics,
)

solidarity_graph = extract_solidarity_subgraph(graph)
num_comp, l_max, ratio = calculate_component_metrics(
    solidarity_graph, total_social_classes=20
)

calculate_liquidity

calculate_liquidity(G)

Calculate potential vs actual solidarity metrics.

Parameters:: G (nx.DiGraph[str]) – Directed graph from WorldState.to_graph()
Returns:: Tuple of (potential_liquidity, actual_liquidity)
Return type:: tuple[int, int]

Interpretation:

potential: Sympathizer network (edges > 0.1)
actual: Cadre network (edges > 0.5)
If potential > actual * 2: Movement is broad but brittle

check_resilience

check_resilience(G, removal_rate=0.2, survival_threshold=0.4, seed=None)

Test if solidarity network survives targeted node removal.

Simulates a “purge” scenario by removing a percentage of nodes and checking if the giant component survives.

Parameters:

G (nx.DiGraph[str]) – Directed graph from WorldState.to_graph()
removal_rate (float) – Fraction of nodes to remove (default 0.2)
survival_threshold (float) – Required survival fraction (default 0.4)
seed (int | None) – RNG seed for reproducibility

Returns:

Result with is_resilient flag and metrics

Return type:

ResilienceResult

Example:

from babylon.engine.topology_monitor import check_resilience

result = check_resilience(graph, removal_rate=0.2, seed=42)
if not result.is_resilient:
    print("Sword of Damocles: Network is fragile!")

TopologyMonitor Class

class TopologyMonitor(resilience_test_interval=5, resilience_removal_rate=0.2, logger=None)

Observer tracking solidarity network condensation via percolation theory.

Implements SimulationObserver protocol.

Parameters:

resilience_test_interval (int) – Run resilience test every N ticks (0 = disabled)
resilience_removal_rate (float) – Fraction of nodes to remove in test
logger (logging.Logger | None) – Logger instance (default: module logger)

Configuration

Parameter	Default	Description
`resilience_test_interval`	5	Test resilience every N ticks (0 disables)
`resilience_removal_rate`	0.2	Fraction of nodes removed in purge test

Properties

name: str: Returns "TopologyMonitor"

history: list[TopologySnapshot]: Returns copy of recorded snapshots (one per tick)

Lifecycle Hooks

on_simulation_start(initial_state, config): Called when simulation begins. Clears history and records initial snapshot.

on_tick(previous_state, new_state): Called after each tick. Records snapshot and logs narrative states.

on_simulation_end(final_state): Called when simulation ends. Logs summary statistics.

Event Emission (Sprint 3.3)

The TopologyMonitor emits PhaseTransitionEvent when percolation ratio crosses threshold boundaries. Events are collected and injected into the next tick’s WorldState.

Internal State:

Attribute	Type	Description
`_previous_phase`	`str \| None`	Last known phase state
`_pending_events`	`list[SimulationEvent]`	Events awaiting collection

Methods:

_classify_phase(percolation_ratio, cadre_density=0.0)

Classify current phase state from percolation ratio and cadre density.

Parameters:

percolation_ratio (float) – Current L_max / N ratio
cadre_density (float) – Ratio of actual/potential liquidity (default 0.0)

Returns:

Phase state name

Return type:

str (“gaseous” | “transitional” | “liquid” | “solid”)

get_pending_events()

Return and clear pending events list.

Returns:: List of events awaiting injection
Return type:: list[SimulationEvent]

Phase States (4-Phase Model - Sprint 3.3):

State	Threshold	Political Meaning
Gaseous	`ratio < 0.1`	Atomized leftism, no coordination capacity
Transitional	`0.1 <= ratio < 0.5`	Emerging structure, vulnerable to disruption
Liquid	`ratio >= 0.5 AND cadre < 0.5`	Mass movement formed, broad but lacks discipline
Solid	`ratio >= 0.5 AND cadre >= 0.5`	Vanguard party crystallized, iron discipline

Narrative States

The monitor logs these narrative states based on metrics:

State	Condition	Log Message
Gaseous	`percolation_ratio < 0.1`	“Movement is atomized”
Liquid	`ratio >= 0.5 AND cadre < 0.5`	“Mass movement formed, lacks cadre discipline”
Solid	`ratio >= 0.5 AND cadre >= 0.5`	“Vanguard Party crystallized, iron discipline”
Crystallization	`liquid -> solid` transition	“Mass movement hardened into disciplined vanguard”
Brittle	`potential > actual * 2`	“Movement is broad but brittle”
Sword of Damocles	`is_resilient == False`	“A purge would destroy the movement”

Usage Example

from babylon.engine.simulation import Simulation
from babylon.engine.topology_monitor import TopologyMonitor
from babylon.models import WorldState, SimulationConfig

# Create initial state
state = WorldState(entities={...}, relationships=[...])
config = SimulationConfig()

# Create monitor
monitor = TopologyMonitor(resilience_test_interval=5)

# Create simulation with observer
sim = Simulation(state, config, observers=[monitor])

# Run simulation
for _ in range(10):
    sim.step()

# Access metrics
for snapshot in monitor.history:
    print(f"Tick {snapshot.tick}: p={snapshot.percolation_ratio:.2f}")

sim.end()

Topology System Reference

Overview

Constants & Thresholds

Phase Detection

Liquidity Classification

Resilience Testing

Metrics

TopologySnapshot

ResilienceResult

Functions

extract_solidarity_subgraph

calculate_component_metrics

calculate_liquidity

check_resilience

TopologyMonitor Class

Configuration

Properties

Lifecycle Hooks

Event Emission (Sprint 3.3)

Narrative States

Usage Example

See Also