babylon.kernel

The kernel — Babylon’s bottom layer (Program 14, Constitution II.6).

Framework abstractions every layer may import and that import nothing above themselves at runtime: the event bus, the system base class + protocol, the graph substrate protocol, and the DI services protocol. These are exactly the constructs whose former home inside babylon.engine forced economics, persistence, and formulas to import the engine backward (the cycles broken in Program 14 Phase 1).

Layering law (enforced by import-linter): kernel < models/formulas < domain packages < engine. Annotations may reference upward under TYPE_CHECKING; the runtime import graph may not.

class babylon.kernel.BlockedEvent(event, interceptor_name, reason, blocked_at=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))[source]

Bases: object

Audit record for blocked events.

blocked_at defaults to the deterministic sim-time of the wrapped event’s tick (Constitution III.7), never the wall clock.

Parameters:

event (Event)
interceptor_name (str)
reason (str)
blocked_at (datetime)

event: Event

interceptor_name: str

reason: str

blocked_at: datetime = datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)

__post_init__()[source]

Derive blocked_at from the wrapped event’s tick (III.7).

Return type:: None

__init__(event, interceptor_name, reason, blocked_at=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))

Parameters:

event (Event)
interceptor_name (str)
reason (str)
blocked_at (datetime)

Return type:

None

class babylon.kernel.Event(type, tick, payload, timestamp=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))[source]

Bases: object

Immutable event representing a simulation occurrence.

Events are frozen dataclasses to ensure they cannot be modified after creation, maintaining integrity of the event history.

Parameters:

type (str)
tick (int)
payload (dict[str, Any])
timestamp (datetime)

type: Event type identifier (e.g., “tick”, “rupture”, “synthesis”)

tick: Simulation tick when the event occurred

payload: Event-specific data dictionary

timestamp: Deterministic sim-time derived from tick (Constitution III.7)

type: str

tick: int

payload: dict[str, Any]

timestamp: datetime = datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc)

__post_init__()[source]

Derive the default timestamp from tick (Constitution III.7).

Return type:: None

__init__(type, tick, payload, timestamp=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc))

Parameters:

type (str)
tick (int)
payload (dict[str, Any])
timestamp (datetime)

Return type:

None

class babylon.kernel.EventBus[source]

Bases: object

Publish/subscribe event bus for simulation components.

The EventBus enables decoupled communication between systems. Components can subscribe to specific event types and will be notified when events of that type are published.

All published events are stored in history for replay/debugging.

Epoch 1→2 Bridge: Supports optional interceptor chain for adversarial mechanics. If no interceptors are registered, events flow through with zero overhead (backwards compatible).

The interceptor chain processes events before emission: - Interceptors are sorted by priority (higher runs first) - Each interceptor can ALLOW, BLOCK, or MODIFY the event - If blocked, the event is logged and not emitted - If modified, the modified event continues through the chain

Example

>>> bus = EventBus()
>>> def on_tick(event: Event) -> None:
...     print(f"Tick {event.tick}: {event.payload}")
>>> bus.subscribe("tick", on_tick)
>>> bus.publish(Event(type="tick", tick=1, payload={"value": 42}))
Tick 1: {'value': 42}

__init__()[source]

Initialize an empty event bus.

Return type:: None

subscribe(event_type, handler)[source]

Subscribe a handler to receive events of a specific type.

Parameters:

event_type (str) – The type of events to subscribe to
handler (Callable[[Event], None]) – Callable that receives Event objects

Return type:

None

register_interceptor(interceptor)[source]

Register an interceptor to process events before emission.

Interceptors are sorted by priority (higher first) each time an event is published. Multiple interceptors with the same priority execute in registration order.

Parameters:: interceptor (EventInterceptor) – The interceptor to register.
Return type:: None

Example

>>> from babylon.kernel.interceptor import EventInterceptor
>>> bus = EventBus()
>>> bus.register_interceptor(my_security_interceptor)

unregister_interceptor(interceptor)[source]

Remove an interceptor from the chain.

Parameters:: interceptor (EventInterceptor) – The interceptor to remove.
Raises:: ValueError – If the interceptor is not registered.
Return type:: None

publish(event, context=None)[source]

Publish an event to all subscribed handlers.

If interceptors are registered, the event passes through the interceptor chain first. If any interceptor blocks the event, it is logged to the blocked events audit channel and not emitted.

The event is stored in history only if it passes all interceptors.

Parameters:

event (Event) – The event to publish.
context (WorldContext | None) – Optional world context for interceptors. Required for Epoch 2 adversarial mechanics.

Return type:

None

get_history()[source]

Get a copy of all published events.

Return type:: list[Event]
Returns:: List of events in chronological order (oldest first).

get_blocked_events()[source]

Get a copy of all blocked events.

The blocked events audit channel records every event that was stopped by an interceptor, including the blocking reason.

Return type:: list[BlockedEvent]
Returns:: List of BlockedEvent records in chronological order.

clear_history()[source]

Remove all events from history.

Return type:: None

clear_blocked_events()[source]

Remove all blocked event records.

Return type:: None

property interceptor_count: int: Number of registered interceptors.

class babylon.kernel.EventInterceptor[source]

Bases: ABC

Abstract base for event interceptors.

Priority ranges (higher runs first): - 90-100: Security/State (block first) - 50-89: Faction/adversarial - 10-49: Resource/validation - 1-9: Logging/audit (run last)

abstract property name: str: Interceptor name for logs and audit.

property priority: int: Chain priority (higher = earlier). Default 100.

abstractmethod intercept(event, context)[source]

Process event. Return allow/block/modify result.

Return type:

InterceptResult

Parameters:

event (Event)
context (WorldContext | None)

class babylon.kernel.GraphProtocol(*args, **kwargs)[source]

Bases: Protocol

Protocol for backend-agnostic graph operations.

Systems interact with the simulation graph ONLY through this protocol. The concrete implementation (NetworkX, DuckDB) is hidden behind adapters.

This protocol is runtime_checkable, enabling isinstance() checks for protocol compliance.

Example

>>> class MyAdapter:
...     def add_node(self, node_id: str, node_type: str, **attrs: Any) -> None:
...         pass
...     # ... implement all 16 methods
>>> adapter = MyAdapter()
>>> isinstance(adapter, GraphProtocol)
True

add_node(node_id, node_type, **attributes)[source]

Add a node with type marker and arbitrary attributes.

Parameters:

node_id (str) – Unique identifier for the node.
node_type (str) – Discriminator for polymorphism (e.g., ‘social_class’).
**attributes (Any) – Type-specific attributes to store on the node.

Return type:

None

get_node(node_id)[source]

Retrieve node by ID.

Parameters:: node_id (str) – The node identifier to look up.
Return type:: GraphNode | None
Returns:: GraphNode model if found, None otherwise.

update_node(node_id, **attributes)[source]

Partial update of node attributes (merge, not replace).

Parameters:

node_id (str) – The node identifier to update.
**attributes (Any) – Attributes to update (merged with existing).

Raises:

KeyError – If node does not exist.

Return type:

None

remove_node(node_id)[source]

Remove node and all incident edges.

Parameters:: node_id (str) – The node identifier to remove.
Raises:: KeyError – If node does not exist.
Return type:: None

add_edge(source, target, edge_type, weight=1.0, **attributes)[source]

Add directed edge with type, weight, and attributes.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge category (e.g., ‘SOLIDARITY’, ‘EXPLOITATION’).
weight (float) – Generic weight (default 1.0).
**attributes (Any) – Type-specific attributes to store on the edge.

Return type:

None

get_edge(source, target, edge_type)[source]

Retrieve specific edge by source, target, and type.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.

Return type:

GraphEdge | None

Returns:

GraphEdge model if found, None otherwise.

update_edge(source, target, edge_type, **attributes)[source]

Partial update of edge attributes.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.
**attributes (Any) – Attributes to update (merged with existing).

Raises:

KeyError – If edge does not exist.

Return type:

None

remove_edge(source, target, edge_type)[source]

Remove specific edge.

Parameters:

source (str) – Source node ID.
target (str) – Target node ID.
edge_type (str) – Edge type to match.

Raises:

KeyError – If edge does not exist.

Return type:

None

get_neighborhood(node_id, radius=1, edge_types=None, direction='out')[source]

Get all nodes within radius hops of the source node.

Parameters:

node_id (str) – Center node for neighborhood.
radius (int) – Maximum hop distance (1 = immediate neighbors).
edge_types (set[str] | None) – Filter to specific edge types (None = all).
direction (Literal['out', 'in', 'both']) – Which edges to follow: outgoing, incoming, or both.

Return type:

Any

Returns:

SubgraphView or equivalent containing nodes in neighborhood.

Raises:

KeyError – If node does not exist.

execute_traversal(query)[source]

Execute a generic traversal query.

This is the hook for complex operations like percolation analysis, pathfinding, and component detection.

Parameters:: query (TraversalQuery) – TraversalQuery specifying the traversal.
Return type:: TraversalResult
Returns:: TraversalResult with nodes, edges, paths, or aggregates.
Raises:: ValueError – If query_type is not supported.

shortest_path(source, target, edge_types=None, weight_attr=None)[source]

Find shortest path between two nodes.

Parameters:

source (str) – Start node ID.
target (str) – End node ID.
edge_types (set[str] | None) – Filter to specific edge types.
weight_attr (str | None) – Attribute to use as weight (None = hop count).

Return type:

list[str] | None

Returns:

List of node IDs in path, or None if no path exists.

query_nodes(node_type=None, predicate=None, attributes=None)[source]

Query nodes with optional filtering.

Returns an iterator for DuckDB compatibility (lazy evaluation).

Parameters:

node_type (str | None) – Filter by node type (None = all types).
predicate (Callable[[GraphNode], bool] | None) – Python callable for complex filtering.
attributes (dict[str, Any] | None) – Attribute equality filter (DuckDB-translatable).

Return type:

Iterator[GraphNode]

Returns:

Iterator of matching GraphNode models.

query_edges(edge_type=None, predicate=None, min_weight=None, max_weight=None)[source]

Query edges with optional filtering.

Returns an iterator for DuckDB compatibility (lazy evaluation).

Parameters:

edge_type (str | None) – Filter by edge type.
predicate (Callable[[GraphEdge], bool] | None) – Python callable for complex filtering.
min_weight (float | None) – Minimum weight threshold.
max_weight (float | None) – Maximum weight threshold.

Return type:

Iterator[GraphEdge]

Returns:

Iterator of matching GraphEdge models.

count_nodes(node_type=None)[source]

Count nodes, optionally by type.

Parameters:: node_type (str | None) – Filter by node type (None = count all).
Return type:: int
Returns:: Number of matching nodes.

count_edges(edge_type=None)[source]

Count edges, optionally by type.

Parameters:: edge_type (str | None) – Filter by edge type (None = count all).
Return type:: int
Returns:: Number of matching edges.

aggregate(target, group_by=None, agg_func='count', agg_attr=None)[source]

Aggregate over nodes or edges.

Parameters:

target (Literal['nodes', 'edges']) – Whether to aggregate nodes or edges.
group_by (str | None) – Attribute to group by (e.g., ‘type’).
agg_func (Literal['count', 'sum', 'avg', 'min', 'max']) – Aggregation function to apply.
agg_attr (str | None) – Attribute to aggregate (required for sum/avg/min/max).

Return type:

dict[str, float]

Returns:

Dict mapping group keys to aggregated values.

Example

>>> graph.aggregate("nodes", group_by="type")
{"social_class": 4, "territory": 2}

get_graph_attr(key, default=None)[source]

Retrieve a graph-level attribute.

Graph attributes store global metadata (e.g., economy state, base_year, tick_dynamics). Maps to a metadata table in DuckDB.

Parameters:

key (str) – Attribute name to retrieve.
default (Any) – Value to return if attribute not present.

Return type:

Any

Returns:

The attribute value or default.

set_graph_attr(key, value)[source]

Set a graph-level attribute.

Parameters:

key (str) – Attribute name to set.
value (Any) – Value to store.

Return type:

None

query_faction_influence_by_territory(territory_id)[source]

Return all INFLUENCES edges pointing at a Territory.

Spec-070 FR-021 winning-faction resolution. Each row is (faction_id, influence_level, support_type). Sorted by influence_level descending, lex-ID ascending on ties.

Parameters:: territory_id (str) – Target Territory node ID.
Return type:: list[tuple[str, float, str]]
Returns:: Deterministic list of influencing factions; empty if none.

query_sovereign_claims(sovereign_id)[source]

Return all CLAIMS edges originating from a Sovereign.

Each row is (territory_id, control_level, legal_status). Sorted by control_level descending, lex-ID ascending on ties. Used by SovereigntySystem + CollapseTransitionSystem.

Parameters:: sovereign_id (str) – Sovereign node ID.
Return type:: list[tuple[str, float, str]]
Returns:: Deterministic list of claims; empty if none.

query_territory_claims(territory_id)[source]

Return all CLAIMS edges pointing at a Territory.

Each row is (sovereign_id, control_level, legal_status). Sorted by control_level descending, lex-ID ascending on ties. Used by SovereigntySystem for dual-power detection (FR-035) and effective-controller resolution (FR-020).

Parameters:: territory_id (str) – Target Territory node ID.
Return type:: list[tuple[str, float, str]]
Returns:: Deterministic list of claimants; empty if none.

query_adjacent_territories(territory_id)[source]

Return sorted list of Territory IDs adjacent via EdgeType.ADJACENCY.

ADJACENCY edges are conceptually bidirectional; this method abstracts the in/out edge direction. Output is sorted lexicographically for determinism.

Parameters:: territory_id (str) – Anchor Territory node ID.
Return type:: list[str]
Returns:: Sorted list of adjacent Territory IDs; empty if isolated.

bulk_partition_claims(from_sovereign_id, to_sovereign_id, territories)[source]

Atomically rewire CLAIMS edges from one Sovereign to another for the given Territory set (spec-070 FR-027).

Performance requirement (FR-018 / SC-004): MUST be implementable in O(K) where K = len(territories) — NOT O(N) in the unchanged-territory count.

Parameters:

from_sovereign_id (str) – Current owner Sovereign.
to_sovereign_id (str) – New owner Sovereign.
territories (set[str]) – Set of Territory IDs to migrate.

Return type:

int

Returns:

Count of edges actually rewired.

query_contiguous_component_under_predicate(territory_seed, predicate)[source]

BFS over EdgeType.ADJACENCY from territory_seed, collecting Territories satisfying predicate.

Bounded by the predicate-satisfying contiguous region size, not global graph size (per FR-018).

Parameters:

territory_seed (str) – BFS start node.
predicate (Callable[[str], bool]) – (territory_id) -> bool; only matching nodes are included and traversed through.

Return type:

set[str]

Returns:

Set of Territory IDs in the contiguous predicate-satisfying component containing territory_seed (empty if the seed itself fails the predicate).

__init__(*args, **kwargs)

class babylon.kernel.InterceptResult(event, reason='')[source]

Bases: object

Immutable result of interceptor processing.

Parameters:

event (Event | None)
reason (str)

event: Event to continue with, or None if blocked.

reason: Narrative explanation (required if blocked).

event: Event | None

reason: str = ''

__post_init__()[source]

Validate blocked events have a reason.

Return type:: None

classmethod allow(event)[source]

Allow event unchanged.

Return type:: InterceptResult
Parameters:: event (Event)

classmethod block(reason)[source]

Block event with narrative reason.

Return type:: InterceptResult
Parameters:: reason (str)

classmethod modify(new_event, reason='')[source]

Modify event, optionally with reason.

Return type:

InterceptResult

Parameters:

new_event (Event)
reason (str)

property is_blocked: bool: True if event was blocked.

property is_modified: bool: True if event was modified with reason.

__init__(event, reason='')

Parameters:

event (Event | None)
reason (str)

Return type:

None

class babylon.kernel.ServicesProtocol(*args, **kwargs)[source]

Bases: Protocol

Attribute surface of the simulation service container.

Variables:

config – Run-scoped SimulationConfig (carries rng_seed).
database – Database connection satisfying the persistence protocol.
event_bus – Kernel publish/subscribe bus.
formulas – FormulaRegistry of hot-swappable formulas.
defines – GameDefines — the moddable coefficient space.
metrics – Telemetry collector.

config: Any

database: Any

event_bus: EventBus

formulas: Any

defines: Any

metrics: Any

field_registry: Any

opposition_registry: Any

reserve_army_data_source: Any

dispossession_data_source: Any

productivity_data_source: Any

melt_calculator: Any

basket_calculator: Any

gamma_calculator: Any

capital_calculator: Any

throughput_calculator: Any

transition_engine: Any

tensor_registry: Any

economics_fallbacks: Any

community_hypergraph: Any

turnover_profile_source: Any

inventory_data_source: Any

depreciation_data_source: Any

hex_grid: Any

persistence: Any

tracer: Any

boundary_register: Any

auditor: Any

distribution_calculator: Any

interest_calculator: Any

credit_cycle_detector: Any

fictitious_capital_calculator: Any

rent_calculator: Any

housing_calculator: Any

counter_tendency_calculator: Any

value_basis_converter: Any

financial_crisis_assessor: Any

z1_source: Any

housing_data_source: Any

periphery_labor_source: Any

final_demand_source: Any

industry_county_allocator: Any

production_chain_calculator: Any

bea_industries: list[str] | None

__init__(*args, **kwargs)

class babylon.kernel.System(*args, **kwargs)[source]

Bases: Protocol

Protocol defining a historical materialist system.

property name: str: The identifier of the system.

step(graph, services, context)[source]

Apply system logic to the world graph.

Parameters:

graph (GraphProtocol) – Mutable NetworkX graph representing WorldState.
services (ServicesProtocol) – ServicesProtocol with config, formulas, event_bus, database.
context (Union[dict[str, Any], TickContext]) – TickContext or dict with ‘tick’ (int) and optional metadata. TickContext is the preferred type; dict is supported for backward compatibility with existing tests.

Return type:

None

__init__(*args, **kwargs)

class babylon.kernel.SystemBase[source]

Bases: ABC

Abstract base for all simulation Systems.

Subclasses MUST set the name ClassVar and implement step().

Helpers:

_wrap_graph() — assert the graph satisfies GraphProtocol. _read() — read a node attribute, raising KeyError when

required=True and the attribute is absent (surfaces schema bugs at the read site, per CLAUDE.md “Common Gotchas”).

_publish() — publish an event via the service container’s bus.

name: ClassVar[str]

creates_value: ClassVar[bool] = False

abstractmethod step(graph, services, context)[source]

Apply system logic to the world graph (in-place mutation).

The signature matches babylon.kernel.system_protocol.System exactly. The engine (and all test fixtures) pass a BabylonGraph, which satisfies GraphProtocol.

Return type:

None

Parameters:

graph (GraphProtocol)
services (ServicesProtocol)
context (ContextType)

babylon.kernel.resolve_rng(services, tick)[source]

Seed-deterministic RNG for stochastic System rolls (III.7).

Prefers services.rng when a harness injects one; otherwise a fresh random.Random(0xBA1AC1A + tick) — the spec-070 fallback previously duplicated by faction_influence and reactionary.

Parameters:

services (ServicesProtocol) – Service container (checked for an injected rng).
tick (int) – Current simulation tick, mixed into the fallback seed.

Return type:

Random

Returns:

A random.Random whose stream is a pure function of tick (fallback path) or the injected harness RNG.

Modules

`database`	Kernel protocol for the injectable database connection.
`event_bus`	Event system for decoupled communication in the simulation.
`exceptions`	Exception hierarchy for Babylon/Babylon.
`graph_protocol`	Graph Protocol definition for backend-agnostic graph operations.
`interceptor`	Event Interceptor pattern for Epoch 2 adversarial mechanics.
`log`	Custom logging utilities for Babylon.
`math`	Mathematical utilities for simulation precision.
`metrics`	Metrics collector interfaces and protocols.
`retry`	Retry decorator for transient failure handling.
`schema_registry`	Shared JSON Schema registry builder for `$ref` resolution.
`services`	Structural protocol for the simulation's DI service container.
`sim_clock`	Deterministic simulation clock (Constitution III.7).
`system_base`	SystemBase — abstract base class for simulation Systems (ADR-003).
`system_protocol`	Protocol definition for simulation systems.