babylon.engine.history

History Stack module for the Babylon simulation engine.

This module provides: - Pydantic models for history tracking and checkpointing - Stack operations for undo/redo functionality - File I/O for state persistence - Auto-checkpointing for periodic saves

The historical record must survive state repression.

class babylon.engine.history.CheckpointMetadata(**data)[source]

Bases: BaseModel

Metadata for a saved checkpoint.

Captures context about when and why a checkpoint was created. Used for checkpoint listing and identification.

Parameters:

created_at (datetime)
tick (int)
description (str)
version (str)

created_at: UTC timestamp when the checkpoint was created.

tick: The simulation tick at checkpoint time.

description: Optional human-readable description of the checkpoint.

version: Schema version for compatibility checking.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

created_at: datetime

tick: int

description: str

version: str

class babylon.engine.history.Checkpoint(**data)[source]

Bases: BaseModel

Complete checkpoint bundle for save/load.

Contains everything needed to fully restore simulation state: - Metadata for identification and context - WorldState with all entities and relationships - SimulationConfig with all formula coefficients

Parameters:

metadata (CheckpointMetadata)
state (WorldState)
config (SimulationConfig)

metadata: Checkpoint identification and context.

state: The WorldState snapshot.

config: The SimulationConfig active at checkpoint time.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

metadata: CheckpointMetadata

state: WorldState

config: SimulationConfig

class babylon.engine.history.HistoryEntry(**data)[source]

Bases: BaseModel

Single entry in the history stack.

Represents one state snapshot with its tick number. Used by the history stack for undo/redo operations.

Parameters:

tick (int)
state (WorldState)

tick: The simulation tick for this state.

state: The WorldState snapshot.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

tick: int

state: WorldState

class babylon.engine.history.HistoryStack(**data)[source]

Bases: BaseModel

Immutable history stack for undo/redo operations.

Maintains a list of history entries with a current position indicator. All operations return new HistoryStack instances (functional pattern).

The stack supports: - Linear timeline (push after undo truncates future) - Protected ticks that survive pruning - Configurable maximum depth

Parameters:

entries (list[HistoryEntry])
current_index (int)
max_depth (int)
protected_ticks (frozenset[int])

entries: List of history entries (oldest first).

current_index: Index of current state (-1 if empty).

max_depth: Maximum number of entries to retain.

protected_ticks: Tick numbers that cannot be pruned.

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

entries: list[HistoryEntry]

current_index: int

max_depth: int

protected_ticks: frozenset[int]

class babylon.engine.history.CheckpointConfig(**data)[source]

Bases: BaseModel

Configuration for auto-checkpointing behavior.

Controls when and where checkpoints are automatically created.

Parameters:

enabled (bool)
interval (int)
checkpoint_dir (str)
max_checkpoints (int)

enabled: Whether auto-checkpointing is enabled.

interval: Number of ticks between auto-checkpoints.

checkpoint_dir: Directory path for checkpoint files.

max_checkpoints: Maximum checkpoints to retain (0 = unlimited).

model_config: ClassVar[ConfigDict] = {'frozen': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

enabled: bool

interval: int

checkpoint_dir: str

max_checkpoints: int

babylon.engine.history.push_state(history, state)[source]

Add a new state to the history stack.

Behavior: - If current_index is not at the end, truncates future entries (linear timeline) - Enforces max_depth by pruning oldest non-protected entries - Returns new stack with state added

Parameters:

history (HistoryStack) – The current history stack.
state (WorldState) – The WorldState to add.

Return type:

HistoryStack

Returns:

New HistoryStack with the state added.

babylon.engine.history.undo(history)[source]

Move back one state in history.

Behavior: - Returns the previous state (one before current) - Preserves all entries for potential redo - Returns None if at start or empty

Parameters:: history (HistoryStack) – The current history stack.
Return type:: tuple[HistoryStack, WorldState | None]
Returns:: Tuple of (new stack, previous state or None).

babylon.engine.history.redo(history)[source]

Move forward one state in history.

Behavior: - Returns the next state (one after current) - Returns None if at end or empty

Parameters:: history (HistoryStack) – The current history stack.
Return type:: tuple[HistoryStack, WorldState | None]
Returns:: Tuple of (new stack, next state or None).

babylon.engine.history.get_current_state(history)[source]

Get the state at the current position.

Parameters:: history (HistoryStack) – The history stack.
Return type:: WorldState | None
Returns:: Current WorldState or None if stack is empty.

babylon.engine.history.get_state_at_tick(history, tick)[source]

Get the state for a specific tick.

Parameters:

history (HistoryStack) – The history stack.
tick (int) – The tick number to find.

Return type:

WorldState | None

Returns:

WorldState at that tick or None if not found.

babylon.engine.history.prune_history(history, keep_count)[source]

Remove old entries, keeping the most recent ones.

Behavior: - Keeps the most recent keep_count entries - Protected ticks are never pruned (even if that exceeds keep_count) - Adjusts current_index to point to the same tick

Parameters:

history (HistoryStack) – The history stack.
keep_count (int) – Number of recent entries to keep.

Return type:

HistoryStack

Returns:

New HistoryStack with pruned entries.

babylon.engine.history.protect_tick(history, tick)[source]

Mark a tick as protected (cannot be pruned).

Parameters:

history (HistoryStack) – The history stack.
tick (int) – The tick number to protect.

Return type:

HistoryStack

Returns:

New HistoryStack with the tick protected.

exception babylon.engine.history.CheckpointIOError(message)[source]

Bases: StorageError

Base exception for checkpoint I/O errors.

Now inherits from StorageError to be part of unified BabylonError hierarchy.

Parameters:: message (str)
Return type:: None

__init__(message)[source]

Parameters:: message (str)
Return type:: None

exception babylon.engine.history.CheckpointNotFoundError(message)[source]

Bases: CheckpointIOError

Raised when a checkpoint file is not found.

Parameters:: message (str)
Return type:: None

__init__(message)[source]

Parameters:: message (str)
Return type:: None

exception babylon.engine.history.CheckpointCorruptedError(message)[source]

Bases: CheckpointIOError

Raised when a checkpoint file contains invalid JSON.

Parameters:: message (str)
Return type:: None

__init__(message)[source]

Parameters:: message (str)
Return type:: None

exception babylon.engine.history.CheckpointSchemaError(message)[source]

Bases: CheckpointIOError

Raised when a checkpoint file has invalid schema.

Parameters:: message (str)
Return type:: None

__init__(message)[source]

Parameters:: message (str)
Return type:: None

babylon.engine.history.save_state(state, path)[source]

Save a WorldState to a JSON file.

Uses atomic write pattern (temp file + rename) for safety. Creates parent directories if they don’t exist.

Parameters:

state (WorldState) – The WorldState to save.
path (Path) – The file path to save to.

Return type:

Path

Returns:

The path where the file was saved.

babylon.engine.history.load_state(path)[source]

Load a WorldState from a JSON file.

Parameters:

path (Path) – The file path to load from.

Return type:

WorldState

Returns:

The loaded WorldState.

Raises:

CheckpointNotFoundError – If the file doesn’t exist.
CheckpointCorruptedError – If the file contains invalid JSON.
CheckpointSchemaError – If the JSON doesn’t match WorldState schema.

babylon.engine.history.save_checkpoint(state, config, path, description='')[source]

Save a full Checkpoint to a JSON file.

Creates a Checkpoint with current metadata and saves it atomically. Creates parent directories if they don’t exist.

Parameters:

state (WorldState) – The WorldState to save.
config (SimulationConfig) – The SimulationConfig to save.
path (Path) – The file path to save to.
description (str) – Optional description for the checkpoint.

Return type:

Path

Returns:

The path where the file was saved.

babylon.engine.history.load_checkpoint(path)[source]

Load a full Checkpoint from a JSON file.

Parameters:

path (Path) – The file path to load from.

Return type:

Checkpoint

Returns:

The loaded Checkpoint.

Raises:

CheckpointNotFoundError – If the file doesn’t exist.
CheckpointCorruptedError – If the file contains invalid JSON.
CheckpointSchemaError – If the JSON doesn’t match Checkpoint schema.

babylon.engine.history.list_checkpoints(directory)[source]

List all valid checkpoints in a directory.

Scans the directory for JSON files that are valid checkpoints. Returns a list of (path, metadata) tuples for valid checkpoints.

Parameters:: directory (Path) – The directory to scan.
Return type:: list[tuple[Path, CheckpointMetadata]]
Returns:: List of (Path, CheckpointMetadata) tuples for valid checkpoints.

babylon.engine.history.validate_checkpoint_file(path)[source]

Check if a file is a valid checkpoint.

Performs a quick validation without loading the full checkpoint.

Parameters:: path (Path) – The file path to validate.
Return type:: bool
Returns:: True if the file is a valid checkpoint, False otherwise.

class babylon.engine.history.AutoCheckpointer(config, base_dir=None)[source]

Bases: object

Stateful checkpoint manager for automatic periodic saves.

AutoCheckpointer integrates with the game loop to create checkpoints at configurable intervals without modifying the pure step function.

Parameters:

config (CheckpointConfig)
base_dir (Path | None)

config: The CheckpointConfig controlling behavior.

checkpoint_dir: The directory where checkpoints are saved.

__init__(config, base_dir=None)[source]

Initialize the auto-checkpointer.

Parameters:

config (CheckpointConfig) – Configuration for checkpointing behavior.
base_dir (Path | None) – Base directory for checkpoint storage. Defaults to current dir.

Return type:

None

property checkpoint_dir: Path: Get the checkpoint directory path.

force_checkpoint(state, sim_config, description='')[source]

Force creation of a checkpoint regardless of interval.

Useful for manual saves or important milestones.

Parameters:

state (WorldState) – The WorldState to checkpoint.
sim_config (SimulationConfig) – The SimulationConfig to include.
description (str) – Optional description for the checkpoint.

Return type:

Path

Returns:

Path to the created checkpoint.

get_latest_checkpoint()[source]

Get the path to the most recent checkpoint.

Return type:: Path | None
Returns:: Path to the newest checkpoint, or None if no checkpoints exist.

on_step(state, sim_config)[source]

Check if a checkpoint should be created after a step.

Called after each simulation step. Creates a checkpoint if: - Auto-checkpointing is enabled - The current tick is a multiple of the interval (including 0)

Parameters:

state (WorldState) – The current WorldState after the step.
sim_config (SimulationConfig) – The SimulationConfig in use.

Return type:

Path | None

Returns:

Path to the created checkpoint, or None if no checkpoint was created.

rotate_checkpoints()[source]

Remove old checkpoints to stay within max_checkpoints limit.

Removes the oldest checkpoints when the count exceeds max_checkpoints. If max_checkpoints is 0 (unlimited), no rotation occurs.

Return type:: list[Path]
Returns:: List of paths that were removed.

babylon.engine.history.create_checkpointed_step(sim_config, checkpointer)[source]

Create a wrapper function that checkpoints after each step.

This enables integration of checkpointing with the pure game loop without modifying the step function itself.

Usage:

checkpointer = AutoCheckpointer(checkpoint_config) checkpointed_step = create_checkpointed_step(sim_config, checkpointer)

for _ in range(100):: state, checkpoint_path = checkpointed_step(state) # checkpoint_path is None unless a checkpoint was created

Parameters:

sim_config (SimulationConfig) – The SimulationConfig for checkpoints.
checkpointer (AutoCheckpointer) – The AutoCheckpointer instance.

Return type:

Callable[[WorldState], tuple[WorldState, Path | None]]

Returns:

A callable that takes WorldState and returns (WorldState, Path | None).

Modules

`auto_checkpoint`	Auto-checkpointing for the History Stack system.
`io`	File I/O operations for the History Stack system.
`models`	Pydantic models for the History Stack system.
`stack`	Pure stack operations for the History Stack system.