babylon.engine.observers.causal

CausalChainObserver for detecting the “Shock Doctrine” pattern (Sprint 3.2).

The CausalChainObserver is a SimulationObserver that detects causal chains in simulation state changes and outputs structured JSON NarrativeFrame.

Theoretical Context (Naomi Klein’s “Shock Doctrine”):

The Shock Doctrine pattern describes how economic crises (shocks) are exploited to impose austerity measures while populations are disoriented. This leads to radicalization as material conditions worsen:

1. ECONOMIC_SHOCK: Pool drops > 20% (crisis hits)

2. AUSTERITY_RESPONSE: Wages decrease (bourgeoisie cuts costs)

3. RADICALIZATION: P(Revolution) increases (class consciousness rises)

Detection Logic:

• Maintain rolling 5-tick history buffer

• Check for sequential pattern: Crash(N) -> Austerity(N+1) -> Radicalization(N+2)

• Output JSON NarrativeFrame with causal graph structure

Output Format:

[NARRATIVE_JSON] {“causal_graph”: {“nodes”: […], “edges”: […]}}

Classes

CausalChainObserver([logger])	Observer detecting the Shock Doctrine causal chain pattern.
TickSnapshot(tick, pool, wage, p_rev)	Immutable snapshot of economic metrics at a single tick.

class babylon.engine.observers.causal.TickSnapshot(tick, pool, wage, p_rev)

Bases: NamedTuple

Immutable snapshot of economic metrics at a single tick.

Parameters:

• tick (int)

• pool (float)

• wage (float)

• p_rev (float)

tick

The tick number.

pool

Imperial rent pool value (Currency).

wage

Current super-wage rate (Coefficient).

p_rev

Maximum P(Revolution) across all entities.

tick: int

Alias for field number 0

pool: float

Alias for field number 1

wage: float

Alias for field number 2

p_rev: float

Alias for field number 3

class babylon.engine.observers.causal.CausalChainObserver(logger=None)

Bases: object

Observer detecting the Shock Doctrine causal chain pattern.

Implements SimulationObserver protocol to receive state change notifications and analyze for the Crash -> Austerity -> Radicalization pattern that emerges from economic crises.

The pattern is detected when: - Tick N: Pool drops >= 20% (ECONOMIC_SHOCK) - Tick N+1 or later: Wage decreases (AUSTERITY_RESPONSE) - Tick N+2 or later: P(Revolution) increases (RADICALIZATION)

When detected, outputs a JSON NarrativeFrame with [NARRATIVE_JSON] prefix at WARNING level for AI narrative generation.

Parameters:

logger (logging.Logger | None)

CRASH_THRESHOLD

Class constant defining crash trigger (-0.20 = 20% drop).

BUFFER_SIZE

Size of the rolling history buffer (5 ticks).

name

Observer identifier (“CausalChainObserver”).

Example

>>> from babylon.engine.observers.causal import CausalChainObserver
>>> observer = CausalChainObserver()
>>> observer.name
'CausalChainObserver'

CRASH_THRESHOLD: float = -0.2

Percentage drop threshold that triggers economic shock detection (-20%).

BUFFER_SIZE: int = 5

Size of the rolling history buffer for pattern detection.

__init__(logger=None)

Initialize CausalChainObserver.

Parameters:

logger (Logger | None) – Logger instance for narrative JSON output. Defaults to module-level logger if not provided.

Return type:

None

property name: str

Return observer identifier.

Returns:

String “CausalChainObserver” for logging and debugging.

on_simulation_start(initial_state, config)

Called when simulation begins.

Clears the history buffer and records the initial state as baseline.

Parameters:

• initial_state (WorldState) – WorldState at tick 0.

• config (SimulationConfig) – SimulationConfig for this run (unused).

Return type:

None

on_tick(previous_state, new_state)

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

Records the new state and checks for the Shock Doctrine pattern. If detected, outputs a JSON NarrativeFrame.

Parameters:

• previous_state (WorldState) – WorldState before the tick (unused, history is internal).

• new_state (WorldState) – WorldState after the tick.

Return type:

None

on_simulation_end(final_state)

Called when simulation ends.

No-op for CausalChainObserver. No cleanup or summary needed.

Parameters:

final_state (WorldState) – Final WorldState when simulation ends (unused).

Return type:

None