babylon.engine.observers

Observer implementations for simulation state monitoring.

This package contains concrete SimulationObserver implementations that monitor specific aspects of the simulation state:

EconomyMonitor: Detects sudden drops in imperial_rent_pool (>20%)
CausalChainObserver: Detects Shock Doctrine pattern (Crash -> Austerity -> Radicalization)

Observers follow the Observer Pattern: they receive state change notifications but cannot modify simulation state. This separation allows AI components to generate narrative from state changes without affecting the deterministic mechanics.

Schema validation is provided for observer JSON outputs:

validate_narrative_frame: Validate NarrativeFrame against JSON schema
is_valid_narrative_frame: Boolean check for NarrativeFrame validity

class babylon.engine.observers.CausalChainObserver(logger=None)[source]

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'

BUFFER_SIZE: int = 5: Size of the rolling history buffer for pattern detection.

CRASH_THRESHOLD: float = -0.2: Percentage drop threshold that triggers economic shock detection (-20%).

__init__(logger=None)[source]

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

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

on_simulation_start(initial_state, config)[source]

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

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

class babylon.engine.observers.EconomyMonitor(logger=None)[source]

Bases: object

Observer detecting economic crises via imperial_rent_pool drops.

Implements SimulationObserver protocol to receive state change notifications and analyze economic state for crisis conditions.

A crisis is detected when the imperial_rent_pool drops by 20% or more from the previous tick. The [CRISIS_DETECTED] log marker allows AI narrative systems to respond appropriately.

Parameters:: logger (logging.Logger | None)

CRISIS_THRESHOLD: Class constant defining crisis trigger (-0.20 = 20% drop).

name: Observer identifier (“EconomyMonitor”).

Example

>>> from babylon.engine.observers.economic import EconomyMonitor
>>> monitor = EconomyMonitor()
>>> monitor.name
'EconomyMonitor'

CRISIS_THRESHOLD: float = -0.2: Percentage drop threshold that triggers crisis detection (-20%).

__init__(logger=None)[source]

Initialize EconomyMonitor.

Parameters:: logger (Logger | None) – Logger instance for crisis warnings. Defaults to module-level logger if not provided.
Return type:: None

property name: str

Return observer identifier.

Returns:: String “EconomyMonitor” for logging and debugging.

on_simulation_end(final_state)[source]

Called when simulation ends.

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

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

on_simulation_start(initial_state, config)[source]

Called when simulation begins.

No-op for EconomyMonitor. Crisis detection only operates on state transitions, not initial state.

Parameters:

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

Return type:

None

on_tick(previous_state, new_state)[source]

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

Compares imperial_rent_pool between states and logs a warning if the drop exceeds CRISIS_THRESHOLD (20%).

Parameters:

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

Return type:

None

class babylon.engine.observers.MetricsCollector(mode='interactive', rolling_window=50)[source]

Bases: object

Observer that collects simulation metrics for analysis.

Implements SimulationObserver protocol. Extracts entity and edge metrics at each tick, with optional rolling window for memory efficiency in interactive mode.

Parameters:

mode (Literal['interactive', 'batch'])
rolling_window (int)

__init__(mode='interactive', rolling_window=50)[source]

Initialize the collector.

Parameters:

mode (Literal['interactive', 'batch']) – “interactive” uses rolling window, “batch” keeps all history
rolling_window (int) – Maximum ticks to keep in interactive mode

Return type:

None

export_json(path, defines, config, csv_path=None)[source]

Write JSON metadata to file.

Parameters:

path (Path) – Output path for JSON file
defines (GameDefines) – GameDefines with fundamental parameters
config (SimulationConfig) – SimulationConfig with run settings
csv_path (Path | None) – Optional path to associated CSV time-series file

Return type:

None

property history: list[TickMetrics]: Return metrics history as a list.

property latest: TickMetrics | None: Return most recent tick metrics, or None if empty.

property name: str: Return observer identifier.

on_simulation_end(final_state)[source]

Called when simulation ends. No-op for MetricsCollector.

Return type:: None
Parameters:: final_state (WorldState)

on_simulation_start(initial_state, config)[source]

Called when simulation begins. Clears history and records tick 0.

Return type:

None

Parameters:

initial_state (WorldState)
config (SimulationConfig)

on_tick(previous_state, new_state)[source]

Called after each tick completes. Records new state.

Return type:

None

Parameters:

previous_state (WorldState)
new_state (WorldState)

property summary: SweepSummary | None: Return sweep summary, or None if no data collected.

to_csv_rows()[source]

Export metrics history as list of dicts for CSV output.

Return type:: list[dict[str, Any]]

to_json(defines, config, csv_path=None)[source]

Export run metadata as structured JSON for reproducibility.

Captures the causal DAG hierarchy: - Level 1 (Fundamental): GameDefines parameters - Level 2 (Config): SimulationConfig settings - Level 3 (Emergent): SweepSummary computed from simulation

Parameters:

defines (GameDefines) – GameDefines with fundamental parameters
config (SimulationConfig) – SimulationConfig with run settings
csv_path (Path | None) – Optional path to associated CSV time-series file

Return type:

dict[str, Any]

Returns:

Structured dict ready for JSON serialization

babylon.engine.observers.is_valid_narrative_frame(frame)[source]

Check if a NarrativeFrame is valid against the JSON schema.

Convenience method that returns a boolean instead of error list.

Parameters:: frame (dict[str, Any]) – Dictionary representing the NarrativeFrame to validate.
Return type:: bool
Returns:: True if valid, False otherwise.

Example

>>> frame = {
...     "pattern": "SHOCK_DOCTRINE",
...     "causal_graph": {
...         "nodes": [{"id": "n1", "type": "ECONOMIC_SHOCK", "tick": 0}],
...         "edges": []
...     }
... }
>>> is_valid_narrative_frame(frame)
True

babylon.engine.observers.validate_narrative_frame(frame)[source]

Validate a NarrativeFrame against the JSON schema.

Parameters:: frame (dict[str, Any]) – Dictionary representing the NarrativeFrame to validate.
Return type:: list[str]
Returns:: List of validation error messages. Empty list if valid.

Example

>>> frame = {"pattern": "TEST", "causal_graph": {"nodes": [], "edges": []}}
>>> errors = validate_narrative_frame(frame)
>>> # Returns errors because nodes must have minItems: 1

Modules

`causal`	CausalChainObserver for detecting the "Shock Doctrine" pattern (Sprint 3.2).
`economic`	EconomyMonitor observer for economic crisis detection (Sprint 3.1).
`metrics`	MetricsCollector observer for unified simulation metrics.
`schema_validator`	JSON Schema validation for observer outputs (Sprint 3.2).