babylon.engine.history.io

File I/O operations for the History Stack system.

This module provides save/load functionality for: - WorldState snapshots (simple state persistence) - Full Checkpoint bundles (state + config + metadata)

All writes use atomic patterns (temp file + rename) to prevent corruption.

Sprint C: File I/O for Phase 3 persistence layer.

Functions

list_checkpoints(directory)

List all valid checkpoints in a directory.

load_checkpoint(path)

Load a full Checkpoint from a JSON file.

load_state(path)

Load a WorldState from a JSON file.

save_checkpoint(state, config, path[, ...])

Save a full Checkpoint to a JSON file.

save_state(state, path)

Save a WorldState to a JSON file.

validate_checkpoint_file(path)

Check if a file is a valid checkpoint.

Exceptions

CheckpointCorruptedError(message)

Raised when a checkpoint file contains invalid JSON.

CheckpointIOError(message)

Base exception for checkpoint I/O errors.

CheckpointNotFoundError(message)

Raised when a checkpoint file is not found.

CheckpointSchemaError(message)

Raised when a checkpoint file has invalid schema.

exception babylon.engine.history.io.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.io.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.io.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.io.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.io.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.io.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:
babylon.engine.history.io.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.io.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:
babylon.engine.history.io.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.io.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.