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

Bifurcation Topology Analysis

Consciousness-weighted analysis predicting whether crisis routes to fascism or revolution. For conceptual background, see George Jackson Bifurcation Model.

Key Modules:

babylon.bifurcation - Package with all analysis functions
babylon.bifurcation.analysis - Full orchestrator
babylon.bifurcation.consciousness - Sigmoid weighting
babylon.bifurcation.axis - Per-axis contradiction analysis
babylon.bifurcation.bridges - Community bridge detection
babylon.bifurcation.resilience - Topological resilience metrics
babylon.bifurcation.ceiling - Material solidarity ceiling
babylon.bifurcation.legitimation - Legitimation crisis amplifier
babylon.bifurcation.types - Result types
babylon.engine.bifurcation_monitor - Tick-level observer
babylon.engine.community_state_store - Community state protocol

BifurcationDefines

Configurable parameters in GameDefines.bifurcation.

Consciousness Sigmoid (US1)

Parameter	Default	Description
`consciousness_sigmoid_midpoint`	0.4	CI value at sigmoid inflection. Below-center so breakage cliff catches assimilated communities.
`consciousness_sigmoid_steepness`	10.0	Slope at inflection point. Higher values produce a sharper cliff.
`consciousness_filter_threshold`	0.2	Minimum sigmoid output to include edge in filtered subgraph.

Classification (US5)

Parameter	Default	Description
`indeterminate_dead_zone`	0.2	Score within +/-0.2 of 1.0 threshold classifies as “indeterminate”.
`axis_tendency_epsilon`	0.001	Division guard for cross-solidarity / lateral-antagonism ratio.

Legitimation Amplifier (US7)

Parameter	Default	Description
`legitimation_amplifier_scale`	2.0	At zero legitimation, crisis intensity multiplied by this factor.

Solidarity Ceiling (US6)

Parameter	Default	Description
`wage_ceiling_high_ratio`	10.0	Wage gap ratio at which ceiling reaches its minimum.
`wage_ceiling_low_ratio`	2.0	Wage gap ratio at which ceiling reaches its maximum.
`wage_ceiling_min`	0.3	Minimum solidarity ceiling (at extreme wage gaps).
`wage_ceiling_max`	0.9	Maximum solidarity ceiling (at similar wages).
`shared_exploitation_bonus`	0.2	Ceiling bonus when agents share an exploitation source.

Purge Resilience (US4)

Parameter	Default	Description
`purge_removal_rate`	0.2	Fraction of high-degree nodes removed in purge resilience test.

Result Types

BifurcationResult

Frozen Pydantic model produced by bifurcation_tendency(). One instance per analysis call.

Field	Type	Description
`overall_tendency`	`Literal`	`"revolutionary"`, `"fascist"`, or `"indeterminate"`
`per_axis_tendency`	`dict[str, float]`	Axis ID to tendency ratio (all axes, including inactive)
`cross_line_solidarity_count`	`int`	Raw SOLIDARITY edges crossing any contradiction axis
`within_line_solidarity_count`	`int`	Raw SOLIDARITY edges within the same side of an axis
`lateral_antagonism_count`	`int`	Antagonistic edges within same side (lateral conflict)
`upward_antagonism_count`	`int`	Antagonistic edges from marginalized toward hegemonic
`consciousness_weighted_cross_solidarity`	`float`	Sum of consciousness-weighted cross-line solidarity values
`mean_collective_identity_marginalized`	`float`	Mean CI across marginalized communities [0, 1]
`dominant_tendency_distribution`	`dict[str, float]`	ConsciousnessTendency fractions across marginalized communities
`community_bridge_count`	`int`	Number of communities spanning contradiction axes
`bridge_potential_weighted`	`float`	Sum of infrastructure * sigmoid(CI) for bridge communities
`legitimation_index`	`float`	Population-weighted mean legitimation [0, 1]
`raw_beta_0`	`int`	Connected components (all SOLIDARITY edges)
`raw_beta_1`	`int`	Cycle rank (all SOLIDARITY edges)
`filtered_beta_0`	`int`	Connected components (consciousness-filtered edges only)
`filtered_beta_1`	`int`	Cycle rank (consciousness-filtered edges only)
`resilience_under_targeted_purge`	`float`	Post-purge L_max / pre-purge L_max on filtered subgraph [0, 1]
`equivalence_class_distribution`	`dict[int, int]`	Maps class size to count of structurally equivalent groups
`critical_singletons`	`list[str]`	Articulation point node IDs on filtered subgraph
`critical_cutsets`	`list[frozenset[str]]`	Minimum edge cuts on filtered subgraph (bounded by max_cutset_size=3)
`mean_assimilation_ratio_marginalized`	`float`	Mean `f / (l + f)` across marginalized communities [0, 1]. Default: 0.0. (Feature 034)
`crisis_fragile_edge_count`	`int`	Solidarity edges where effective CI < 0.3 (crisis-fragile). Default: 0. (Feature 034)

BifurcationSnapshot

Wraps BifurcationResult with tick metadata. Stored in BifurcationMonitor._bifurcation_history.

Field	Type	Description
`tick`	`int`	Simulation tick when computed (>= 0)
`result`	`BifurcationResult`	Full analysis result for this tick

AxisTendency

Per-contradiction-axis analysis result.

Field	Type	Description
`axis_id`	`str`	Matches `ContradictionAxis.id` (e.g. `"colonial"`, `"patriarchal"`)
`cross_solidarity_weighted`	`float`	Sum of consciousness-weighted cross-line solidarity (>= 0)
`lateral_antagonism_weighted`	`float`	Sum of antagonistic edge weights on same side (>= 0)
`tendency_ratio`	`float`	cross / (lateral + epsilon); > 1.0 = solidarity-dominant
`cross_edge_count`	`int`	Raw count of cross-line solidarity edges
`lateral_edge_count`	`int`	Raw count of lateral antagonism edges
`upward_edge_count`	`int`	Raw count of upward antagonism edges

BridgeInfo

Community spanning a contradiction axis with weighted potential.

Field	Type	Description
`community_type`	`CommunityType`	Which community (e.g. `DISABLED`, `INCARCERATED`)
`axes_spanned`	`list[str]`	ContradictionAxis IDs this community bridges (min length 1)
`collective_identity`	`float`	Raw CI from CommunityConsciousness [0, 1]
`sigmoid_ci`	`float`	Sigmoid-transformed CI [0, 1]
`infrastructure`	`float`	Community infrastructure from CommunityState [0, 1]
`weighted_potential`	`float`	infrastructure * sigmoid_ci
`member_count`	`int`	Number of agents in this community

SolidarityCeiling

Material constraints on solidarity between two agents.

Field	Type	Description
`base_ceiling`	`float`	From wage gap ratio interpolation [0, 1]
`exploitation_bonus`	`float`	+0.2 if shared exploitation source [0, 0.2]
`community_bonus`	`float`	Bonus from shared community membership (>= 0)
`effective_ceiling`	`float`	Clamped sum of all components [0, 1]
`wage_gap_ratio`	`float`	max(w_a, w_b) / min(w_a, w_b)
`geographically_proximate`	`bool`	Whether agents share ADJACENCY-linked territories

Analysis Functions

bifurcation_tendency

bifurcation_tendency(graph, H, community_states, agent_memberships, defines)

Compute full bifurcation analysis.

Combines per-axis contradiction tendency (weakest-link), community bridge potential (consciousness-weighted), legitimation crisis amplifier, and topological resilience (two-pass Betti numbers).

Parameters:

graph (nx.DiGraph) – Simulation DiGraph with social_class and territory nodes
H (xgi.Hypergraph) – XGI hypergraph for community membership lookup
community_states (dict[CommunityType, CommunityState]) – Community consciousness data
agent_memberships (dict[str, set[CommunityType]]) – Agent ID to community memberships mapping
defines (BifurcationDefines) – Configurable parameters

Returns:

Frozen result with all analysis fields populated

Return type:

BifurcationResult

Example:

from babylon.bifurcation import bifurcation_tendency
from babylon.config.defines import BifurcationDefines

result = bifurcation_tendency(
    graph=sim_graph,
    H=hypergraph,
    community_states=states,
    agent_memberships=memberships,
    defines=BifurcationDefines(),
)
print(result.overall_tendency)  # "revolutionary", "fascist", or "indeterminate"

consciousness_sigmoid

consciousness_sigmoid(collective_identity, midpoint, steepness)

Nonlinear transform with breakage cliff for consciousness weighting.

Logistic sigmoid: 1 / (1 + exp(-steepness * (ci - midpoint))). Exponent clamped to [-500, +500] to prevent overflow.

Parameters:

collective_identity (float) – Raw CI value [0, 1]
midpoint (float) – Sigmoid inflection point
steepness (float) – Slope at inflection

Returns:

Transformed value [0, 1]. Near-zero below midpoint, near-one above.

Return type:

float

WeightedSolidarityResult

Result of consciousness-weighted solidarity computation (Feature 034). Extends the original float return with a crisis-fragile marker.

Field	Type	Description
`weight`	`float`	Consciousness-weighted solidarity value (>= 0)
`crisis_fragile`	`bool`	`True` if effective CI < crisis-fragile threshold (0.3). Default: `False`.

Defined in WeightedSolidarityResult. See Ternary Consciousness Reference for full details.

consciousness_weighted_solidarity

consciousness_weighted_solidarity(source_id, target_id, graph, H, community_states, defines)

Weight a solidarity edge by consciousness of connected communities.

For each agent, finds marginalized community memberships via the hypergraph, computes mean CI, then weights the edge by sigmoid(min(source_ci, target_ci)).

Edges where the effective CI (min of both endpoints) falls below the crisis-fragile threshold (0.3) are marked as crisis-fragile (Feature 034).

Parameters:

source_id (str) – Source agent node ID
target_id (str) – Target agent node ID
graph (nx.DiGraph) – The simulation DiGraph (for edge attribute access)
H (xgi.Hypergraph) – XGI hypergraph (for community membership lookup)
community_states (dict[CommunityType, CommunityState]) – Current community consciousness data
defines (BifurcationDefines) – Configurable parameters (sigmoid midpoint/steepness)

Returns:

Weighted solidarity with crisis-fragile flag

Return type:

WeightedSolidarityResult

crosses_contradiction_axis

crosses_contradiction_axis(source_id, target_id, axis, agent_memberships)

Check if an edge crosses the given contradiction axis.

An edge crosses when one endpoint is on the hegemonic side and the other is on the marginalized side.

Parameters:

source_id (str) – Source agent node ID
target_id (str) – Target agent node ID
axis (ContradictionAxis) – The contradiction axis to check
agent_memberships (dict[str, set[CommunityType]]) – Agent ID to community memberships mapping

Returns:

True if the edge spans hegemonic and marginalized sides

Return type:

bool

classify_edge_antagonism

classify_edge_antagonism(source_id, target_id, graph, axis, agent_memberships)

Classify antagonistic direction of an edge along a contradiction axis.

Only EXPLOITATION, REPRESSION, and COMPETITION edges are classified.

Parameters:

source_id (str) – Source agent node ID
target_id (str) – Target agent node ID
graph (nx.DiGraph) – The simulation DiGraph
axis (ContradictionAxis) – The contradiction axis to classify against
agent_memberships (dict[str, set[CommunityType]]) – Agent ID to community memberships mapping

Returns:

"lateral" (same side), "upward" (marginalized toward hegemonic), "downward" (hegemonic toward marginalized), or "none"

Return type:

str

compute_axis_tendency

compute_axis_tendency(graph, H, axis, community_states, agent_memberships, defines)

Compute solidarity vs antagonism balance along a single contradiction axis.

Sums consciousness-weighted cross-line solidarity and lateral antagonism. Returns tendency_ratio = cross / (lateral + epsilon).

Parameters:

graph (nx.DiGraph) – The simulation DiGraph
H (xgi.Hypergraph) – XGI hypergraph for community membership lookup
axis (ContradictionAxis) – The contradiction axis to analyze
community_states (dict[CommunityType, CommunityState]) – Current community consciousness data
agent_memberships (dict[str, set[CommunityType]]) – Agent ID to community memberships mapping
defines (BifurcationDefines) – Configurable parameters

Returns:

AxisTendency with all counts and weighted totals

Return type:

AxisTendency

detect_bridges

detect_bridges(H, community_states, axes, agent_memberships, defines)

Detect communities spanning contradiction axes, weighted by consciousness.

Iterates INSTITUTIONAL_EXCLUSION communities in the hypergraph. For each, checks if members span both sides of any contradiction axis. Returns weighted bridge potential = infrastructure * sigmoid(CI).

Parameters:

H (xgi.Hypergraph) – XGI hypergraph with communities as hyperedges
community_states (dict[CommunityType, CommunityState]) – Current community consciousness and infrastructure
axes (list[ContradictionAxis]) – Contradiction axes to check spanning against
agent_memberships (dict[str, set[CommunityType]]) – Agent ID to community memberships mapping
defines (BifurcationDefines) – Configurable parameters (sigmoid midpoint/steepness)

Returns:

List of BridgeInfo for each community spanning at least one axis

Return type:

list[BridgeInfo]

compute_solidarity_ceiling

compute_solidarity_ceiling(node_a_id, node_b_id, graph, agent_memberships, defines)

Compute material solidarity ceiling between two agents.

The ceiling is the maximum achievable solidarity given material conditions (wage gap, shared exploitation, shared community membership, geography).

Parameters:

node_a_id (str) – First agent node ID
node_b_id (str) – Second agent node ID
graph (nx.DiGraph) – Simulation graph with wealth and edge attributes
agent_memberships (dict[str, set[CommunityType]]) – Agent ID to community memberships mapping
defines (BifurcationDefines) – Bifurcation coefficient configuration

Returns:

SolidarityCeiling with all computed components

Return type:

SolidarityCeiling

compute_legitimation_amplifier

compute_legitimation_amplifier(graph, defines)

Compute crisis amplifier from population-weighted mean legitimation.

Parameters:

graph (nx.DiGraph) – Simulation graph with territory nodes carrying legitimation_index and population
defines (BifurcationDefines) – Provides legitimation_amplifier_scale

Returns:

Crisis amplifier in [1.0, legitimation_amplifier_scale]. Returns 1.0 if no territories.

Return type:

float

Resilience Functions

compute_betti_numbers

compute_betti_numbers(subgraph)

Compute Betti numbers for an undirected graph.

Parameters:: subgraph (nx.Graph) – Undirected solidarity subgraph
Returns:: Tuple of (beta_0, beta_1) where beta_0 = connected components, beta_1 = |E| - |V| + beta_0
Return type:: tuple[int, int]

compute_equivalence_classes

compute_equivalence_classes(subgraph)

Group nodes by structural equivalence (identical neighbor sets).

Parameters:: subgraph (nx.Graph) – Undirected solidarity subgraph
Returns:: Maps class_size to count of classes with that size
Return type:: dict[int, int]

find_critical_singletons

find_critical_singletons(subgraph)

Find articulation points whose removal disconnects the graph.

Parameters:: subgraph (nx.Graph) – Undirected solidarity subgraph
Returns:: Sorted list of node IDs that are articulation points
Return type:: list[str]

find_critical_cutsets

find_critical_cutsets(subgraph, max_cutset_size=3)

Find minimum edge cuts per connected component, bounded by size.

Parameters:

subgraph (nx.Graph) – Undirected solidarity subgraph
max_cutset_size (int) – Maximum cut size to include (default 3)

Returns:

List of frozensets, each containing node IDs in the minimum edge cut

Return type:

list[frozenset[str]]

compute_purge_resilience

compute_purge_resilience(subgraph, removal_rate, seed=None)

Measure resilience to targeted removal of high-degree nodes.

Removes top-degree nodes at the given rate and compares post-purge to pre-purge largest component size.

Parameters:

subgraph (nx.Graph) – Undirected solidarity subgraph
removal_rate (float) – Fraction of nodes to remove [0, 1]
seed (int | None) – RNG seed for tie-breaking (default None)

Returns:

Post-purge L_max / pre-purge L_max, clamped to [0, 1]. Returns 1.0 for empty graphs.

Return type:

float

BifurcationMonitor

class BifurcationMonitor(community_state_store, defines=None, logger=None)

Monitor tracking bifurcation tendency across simulation ticks.

Records BifurcationSnapshot per tick and emits BifurcationTendencyEvent when the overall tendency changes.

Accepts a CommunityStateStore via dependency injection for community consciousness access.

Parameters:

community_state_store (CommunityStateStore) – Protocol-based access to community states
defines (BifurcationDefines | None) – Configurable bifurcation parameters (default: BifurcationDefines())
logger (logging.Logger | None) – Logger instance (default: module logger)

Properties

bifurcation_history: list[BifurcationSnapshot]: Returns copy of recorded bifurcation snapshots (one per tick).

Methods

record_bifurcation(graph, H, agent_memberships, tick)

Run bifurcation analysis and record snapshot.

Calls bifurcation_tendency() with community states from the injected store, records the resulting BifurcationSnapshot, and emits a BifurcationTendencyEvent if the overall tendency changed since the previous tick.

Parameters:

graph (nx.DiGraph) – Simulation DiGraph with social_class and territory nodes
H (xgi.Hypergraph) – XGI hypergraph for community membership lookup
agent_memberships (dict[str, set[CommunityType]]) – Agent ID to community memberships mapping
tick (int) – Current simulation tick

get_pending_events()

Return and clear pending events. Same pattern as TopologyMonitor.get_pending_events().

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

BifurcationTendencyEvent

Emitted when overall_tendency changes between ticks. Inherits TopologyEvent.

Field	Type	Description
`event_type`	`EventType`	Always `BIFURCATION_TENDENCY_CHANGE`
`percolation_ratio`	`float`	Set to 0.0 (not tracked by bifurcation monitor)
`num_components`	`int`	`filtered_beta_0` from BifurcationResult
`previous_tendency`	`str`	Overall tendency before change
`new_tendency`	`str`	Overall tendency after change
`consciousness_weighted_cross_solidarity`	`float`	Sum of consciousness-weighted cross-line solidarity
`mean_collective_identity_marginalized`	`float`	Mean CI across marginalized communities
`bridge_potential_weighted`	`float`	Sum of infrastructure * sigmoid(CI) for bridges
`legitimation_index`	`float`	Population-weighted mean legitimation

CommunityStateStore Protocol

class CommunityStateStore

Protocol for accessing community consciousness state. Enables dependency injection so BifurcationMonitor is decoupled from concrete state storage.

get_all()

Returns:: Mapping of community type to current community state
Return type:: dict[CommunityType, CommunityState]

class InMemoryCommunityStateStore(states)

Default implementation wrapping a dict.

Parameters:: states (dict[CommunityType, CommunityState]) – Initial community states

Usage Example

from babylon.bifurcation import bifurcation_tendency
from babylon.config.defines import BifurcationDefines
from babylon.engine.bifurcation_monitor import BifurcationMonitor
from babylon.engine.community_state_store import InMemoryCommunityStateStore

# Set up community state store
store = InMemoryCommunityStateStore(community_states)

# Create monitor
monitor = BifurcationMonitor(
    community_state_store=store,
    defines=BifurcationDefines(),
)

# Record each tick
monitor.record_bifurcation(
    graph=sim_graph,
    H=hypergraph,
    agent_memberships=memberships,
    tick=current_tick,
)

# Check for tendency changes
events = monitor.get_pending_events()
for event in events:
    print(f"Tendency changed: {event.previous_tendency} -> {event.new_tendency}")

# Access history
for snapshot in monitor.bifurcation_history:
    result = snapshot.result
    print(f"Tick {snapshot.tick}: {result.overall_tendency}")

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

Bifurcation Topology Analysis

BifurcationDefines

Consciousness Sigmoid (US1)

Classification (US5)

Legitimation Amplifier (US7)

Solidarity Ceiling (US6)

Purge Resilience (US4)

Result Types

BifurcationResult

BifurcationSnapshot

AxisTendency

BridgeInfo

SolidarityCeiling

Analysis Functions

bifurcation_tendency

consciousness_sigmoid

WeightedSolidarityResult

consciousness_weighted_solidarity

crosses_contradiction_axis

classify_edge_antagonism

compute_axis_tendency

detect_bridges

compute_solidarity_ceiling

compute_legitimation_amplifier

Resilience Functions

compute_betti_numbers

compute_equivalence_classes

find_critical_singletons

find_critical_cutsets

compute_purge_resilience

BifurcationMonitor

Properties

Methods

BifurcationTendencyEvent

CommunityStateStore Protocol

Usage Example