Saves and event sourcing¶

`axiom.savestore`¶

axiom.savestore — saves stored separately from the universe.

The source tree (and its compiled cache) is the universe definition; play-throughs live in dedicated databases:

~/AxiomAI/
|-- universes/<name>/...                   definition (source + cache)
`-- saves/<universe key>/save_<uuid>.db    one game (runtime state)

Each save db is self-contained — full schema, with a copy of the definition tables taken at creation. Benefits:

Session and the whole engine work unchanged (a single DB path);
patching the universe does not brick the games (they keep their own definition, resynchronised on open via refresh_definition — in-place, runtime entities and game state survive);
one save = one portable file (trivial export/import).

Backward compatibility: historical saves embedded in the universe .db remain listed and playable as-is (storage=’embedded’). Only new saves are created as separate files.

Zero Qt dependency: pure engine.

exception axiom.savestore.SaveStoreError[source]¶: Error from the separate-save store.

axiom.savestore.assets_dir_for_save(save_id)[source]¶

A save’s illustrations folder (not created if missing).

Parameters:: save_id (str)
Return type:: Path

axiom.savestore.copy_save_assets(src_save_id, dst_save_id)[source]¶

Copy one save’s illustrations to another. Returns the number copied.

Parameters:

src_save_id (str)
dst_save_id (str)

Return type:

int

axiom.savestore.delete_save_assets(save_id)[source]¶

Delete a save’s illustrations folder (no-op if missing).

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

axiom.savestore.truncate_save_assets(save_id, last_kept_turn_id)[source]¶

Purge the turn_<n>.png files with n > last_kept_turn_id (rewind).

Returns the number of deleted files. Non-conforming names are ignored.

Parameters:

save_id (str)
last_kept_turn_id (int)

Return type:

int

axiom.savestore.truncate_assets_in(assets_dir, last_kept_turn_id)[source]¶

Variant of truncate_save_assets on an explicit folder (Session with an injected data_dir).

Parameters:

assets_dir (Path)
last_kept_turn_id (int)

Return type:

int

axiom.savestore.universe_key(universe_db)[source]¶

Stable key of a universe, used to file its saves.

Folder universe: name of the source folder. Flat .db: file stem. Based on the shape of the path only (not on universe.toml existing): the key must stay identical even when the source is momentarily missing/broken, otherwise the saves become unreachable.

Parameters:: universe_db (str | Path)
Return type:: str

axiom.savestore.saves_dir_for(universe_db)[source]¶

A universe’s separate-saves folder (not created if missing).

Parameters:: universe_db (str | Path)
Return type:: Path

axiom.savestore.is_separated_save_db(db_path)[source]¶

True when db_path is a separate save (carries a Save_Meta table).

Parameters:: db_path (str | Path)
Return type:: bool

axiom.savestore.create_save(universe_db, player_name, difficulty, player_persona='')[source]¶

Create a new game in its own saves/<universe>/save_<uuid>.db database.

The universe definition is copied into the save db (self-contained). The link to the universe (db + optional source) is recorded in Save_Meta for resynchronisation on open.

Returns:

A dict with keys save_id and db_path — db_path is the database to hand to Session (and to the engine helpers) to play this game.

Parameters:

universe_db (str | Path)
player_name (str)
difficulty (str)
player_persona (str)

Return type:

dict

axiom.savestore.new_save_container(universe_db)[source]¶

Prepare a blank save db (definition copied + Save_Meta, no Saves row).

Common building block of create_save and of the imports (save-import, save-unpack): the caller then creates/imports its Saves row(s) and calls finalize_save_container.

Parameters:: universe_db (str | Path)
Return type:: Path

axiom.savestore.finalize_save_container(container, save_id)[source]¶

Rename a save container to its real save_id, definitively.

Flushes the WAL into the main file BEFORE the rename (otherwise the -wal/-shm sidecars would stay attached to the old name and the last writes would be lost).

Parameters:

container (Path)
save_id (str)

Return type:

Path

axiom.savestore.list_saves(universe_db)[source]¶

List all the games of a universe, separate and embedded (legacy).

Returns:: List of dicts in the db_helpers.load_saves format, enriched with db_path (the database to open for this save) and storage (‘separated’ | ‘embedded’), sorted by last_updated, most recent first.
Parameters:: universe_db (str | Path)
Return type:: list[dict]

axiom.savestore.resolve_save_db(universe_db, save_id)[source]¶

Return the database containing save_id (separate, or the universe itself).

Parameters:

universe_db (str | Path)
save_id (str)

Return type:

str | None

axiom.savestore.prepare_save_for_play(universe_db, save_id)[source]¶

Resolve a save’s database and resync its definition if the source changed.

For a separate save linked to a folder universe: when the source hash differs from the recorded one, refresh_definition is applied to the save db (in-place: journal, runtime entities and game state intact) and the recorded hash is updated. A missing/broken source is not blocking: the save keeps its embedded definition (it is self-contained).

Returns:

The path of the database to hand to Session, or None for an unknown save.

Parameters:

universe_db (str | Path)
save_id (str)

Return type:

str | None

axiom.savestore.refresh_save_definition(save_db)[source]¶

Resynchronise a separate save’s definition from its universe source.

No-op (False) for an embedded save, a save with no linked source, or one already up to date. A malformed source is ignored (the save stays playable as-is).

Parameters:: save_db (str | Path)
Return type:: bool

axiom.savestore.extract_save(universe_db, save_id)[source]¶

Extract an embedded (legacy) save to its own separate file.

Copies the universe’s current definition + all this save’s runtime rows. The original save stays intact in the universe .db (it is a copy, not a move).

Returns:

The path of the new saves/<universe>/save_<id>.db file.

Parameters:

universe_db (str | Path)
save_id (str)

Return type:

Path

axiom.savestore.pack_save(universe_db, save_id, output_path)[source]¶

Export a save to a .axiomsave archive (zip: self-contained save.db + manifest).

A separate save is zipped as-is; an embedded (legacy) save is first extracted to a self-contained file (a copy — the original stays). The vector memory does not travel (empty on import).

Parameters:

universe_db (str | Path)
save_id (str)
output_path (str | Path)

Return type:

Path

axiom.savestore.unpack_save(archive_path, universe_db, force=False)[source]¶

Import a .axiomsave archive into a universe’s save store.

By default, refuses an archive coming from another universe (different universe_key) — pass force=True to override. When the save_id already exists here, the imported save is re-identified (new uuid) so an existing game is never overwritten.

Returns:

A dict with keys save_id and db_path.

Parameters:

archive_path (str | Path)
universe_db (str | Path)
force (bool)

Return type:

dict

axiom.savestore.duplicate_save(universe_db, save_id, player_name=None)[source]¶

Duplicate a game as-is (full journal preserved).

Separate save: re-identified file copy (new uuid) — the “one save = one file” model is preserved, unlike a fork_save within the same file. Embedded (legacy) save: fork at the last turn in the same database, as before.

Returns:

A dict with keys save_id and db_path.

Parameters:

universe_db (str | Path)
save_id (str)
player_name (str | None)

Return type:

dict

axiom.savestore.delete_save(universe_db, save_id)[source]¶

Delete a game. A separate save whose database becomes empty is removed from disk.

Returns:

True when a save was deleted.

Parameters:

universe_db (str | Path)
save_id (str)

Return type:

bool

axiom.savestore.delete_universe_saves(universe_db)[source]¶

Delete a universe’s separate-saves folder (along with the universe), illustrations included.

Parameters:: universe_db (str | Path)
Return type:: None

`axiom.saves`¶

axiom.saves — save editing (creatable/editable by humans or LLMs).

Design decisions:

D1: the Event_Log journal stays the source of truth (it powers the rewind). Editing happens on top of it: the state is materialised at a point (replay), forked (truncated journal), and an edited state is imported as a new save (“genesis” events at turn 0). Derived data (State_Cache/Snapshots) is never edited directly.
D3: an imported save starts with an empty vector memory (it fills up as you play).

Point selector: by turn (at_turn) or by in-game time in minutes (at_minute, resolved through the Timeline table). Zero Qt dependency.

Editable text format, save_state.toml:

[save]      player_name / difficulty / player_persona
[point]     turn_id / in_game_minutes   (informative at export)
[state.<entity_id>]   stat = "value"    (effective entity state)
[[inventory]]         entity_id / item_id / quantity
[[modifiers]]         entity_id / stat_key / delta / minutes_remaining

exception axiom.saves.SaveError[source]¶: Save editing/reading error.

axiom.saves.resolve_point(db_path, save_id, *, at_turn=None, at_minute=None)[source]¶

Resolve a selector (turn or in-game minute) into a turn_id.

at_turn: used as-is.
at_minute: last turn whose Timeline.in_game_time <= at_minute.
neither: last turn of the save.

Parameters:

db_path (str)
save_id (str)
at_turn (int | None)
at_minute (int | None)

Return type:

int

axiom.saves.materialize_state(db_path, save_id, *, at_turn=None, at_minute=None)[source]¶

Materialise a save’s state at a given point (by replaying the journal).

Per-entity stats = the universe’s base stats (Entity_Stats) overlaid with the state replayed up to the point (logical State_Cache). Inventory and modifiers are the current state (tables that are not event-sourced).

Parameters:

db_path (str)
save_id (str)
at_turn (int | None)
at_minute (int | None)

Return type:

dict[str, Any]

axiom.saves.export_save_state(db_path, save_id, out_path, *, at_turn=None, at_minute=None)[source]¶

Export a save’s materialised state to an editable save_state.toml.

Parameters:

db_path (str)
save_id (str)
out_path (str | Path)
at_turn (int | None)
at_minute (int | None)

Return type:

Path

axiom.saves.import_save_state(db_path, state_path, *, player_name=None)[source]¶

Create a new playable save from a save_state.toml.

Seeds the state through “genesis” events at turn 0 (entity_create + stat_set), then materialises State_Cache, the inventory, the modifiers and a Timeline entry. Empty vector memory. Returns the new save_id.

Parameters:

db_path (str)
state_path (str | Path)
player_name (str | None)

Return type:

str

axiom.saves.apply_correction(db_path, save_id, patch, *, at_turn=None)[source]¶

Apply a correction to an existing save without rewriting the past.

Stat changes become manual_edit events (at the chosen turn, default = last turn) — the journal stays consistent and append-only, the rewind is preserved, and the edit is traceable. Inventory and modifiers (not event-sourced) are written directly.

The patch dict has the shape:

{
    "entities":  {entity_id: {stat_key: "value", ...}},
    "inventory": [{entity_id, item_id, quantity}, ...],
    "modifiers": [{entity_id, stat_key, delta, minutes_remaining}, ...],
}

Returns:

The turn_id the correction was appended at.

Parameters:

db_path (str)
save_id (str)
patch (dict[str, Any])
at_turn (int | None)

Return type:

int

axiom.saves.diff_save_states(before, after)[source]¶

Compute the correction patch between two parsed save_state.toml states.

Supports the “edit the save” flow: the state is exported, the user edits the TOML, and only the differences are appended via apply_correction (otherwise every unchanged stat would become a spurious manual_edit event).

stats: changed or added values (removing a stat does not exist in the correction model — ignored);
inventory: changed/added quantities; a vanished line means quantity 0 (= removal);
modifiers: only new ones are kept (a correction can only add modifiers).

Parameters:

before (dict[str, Any])
after (dict[str, Any])

Return type:

dict[str, Any]

axiom.saves.apply_correction_file(db_path, save_id, patch_path, *, at_turn=None)[source]¶

Load a TOML file (same sections as save_state.toml) and apply it as a correction.

Parameters:

db_path (str)
save_id (str)
patch_path (str | Path)
at_turn (int | None)

Return type:

int

axiom.saves.fork_save(db_path, save_id, *, at_turn=None, at_minute=None, player_name=None)[source]¶

Create a new save = save_id’s journal truncated at the chosen point.

The full journal up to the point is copied (rewind/audit preserved); current inventory and modifiers are copied as-is. Returns the new save_id.

Parameters:

db_path (str)
save_id (str)
at_turn (int | None)
at_minute (int | None)
player_name (str | None)

Return type:

str

`axiom.checkpoint`¶

database/checkpoint.py

Checkpoint and rewind management for Axiom AI saves.

The CheckpointManager exposes the rewind primitive (deleting future events and rebuilding the State_Cache), a save listing helper, and the destructive Hardcore-mode save deletion.

class axiom.checkpoint.CheckpointManager(db_path)[source]¶

Manages save checkpoints, rewinds, and Hardcore deletion for one universe.

Parameters:: db_path (str) – Filesystem path to an existing universe .db file created by database.schema.create_universe_db().

rewind(save_id, target_turn_id)[source]¶

Revert a save to its state at target_turn_id.

Steps:

Count how many future events will be deleted (for the summary).
DELETE all Event_Log rows where turn_id > target_turn_id.
Rebuild State_Cache from the surviving events.

Parameters:

save_id (str) – The save to rewind.
target_turn_id (int) – The turn to revert to (inclusive). All events with turn_id strictly greater than this value are permanently removed.

Returns:

A dict with keys deleted_events and rebuilt_to_turn.

Return type:

A summary dict

Raises:

sqlite3.Error – On any database failure.

list_checkpoints(save_id)[source]¶

Return the distinct turn IDs present in Event_Log for a save, ascending.

This list represents the turns the player could rewind to. The UI can use it to populate a “rewind to turn …” selector.

Parameters:: save_id (str) – The save whose checkpoint turns are requested.
Returns:: Sorted list of unique turn_id integers. Empty list if the save has no recorded events.
Raises:: sqlite3.Error – On any database failure.
Return type:: list[int]

delete_save(save_id, universe_dir)[source]¶

Irrevocably delete a save and its associated universe directory.

Intended exclusively for Hardcore mode upon player death. This method:

Removes the save row from the database (cascades to Event_Log, etc).
Attempts to delete the universe_dir from the filesystem.

Parameters:

save_id (str) – The save to erase from the database.
universe_dir (str) – Absolute path to the universe directory to delete.

Raises:

OSError – If the directory cannot be deleted after multiple retries.
FileNotFoundError – If universe_dir does not exist.
sqlite3.Error – On any database failure.

Return type:

None

`axiom.events`¶

database/event_sourcing.py

Core Event Sourcing implementation for Axiom AI.

Every state change in the game is recorded as an immutable event in Event_Log. The State_Cache is a performance materialisation derived by replaying those events. This module is the authoritative bridge between the two.

Supported event_type values (non-exhaustive; the engine is extensible):

‘entity_create’ payload: {“entity_id”: str, “entity_type”: str, “name”: str}
‘stat_change’ payload: {“entity_id”: str, “stat_key”: str, “delta”: float}
OR {“entity_id”: str, “stat_key”: str, “value”: str}
‘stat_set’ payload: {“entity_id”: str, “stat_key”: str, “value”: str}
‘dialogue’ payload: {“speaker”: str, “text”: str} (no cache mutation)
‘combat_roll’ payload: {“entity_id”: str, …} (no cache mutation)

class axiom.events.EventSourcer(db_path)[source]¶

Manages event appending, querying, and State_Cache reconstruction for a single Axiom AI universe database.

Parameters:: db_path (str) – Filesystem path to an existing universe .db file created by database.schema.create_universe_db().

append_event(save_id, turn_id, event_type, target_entity, payload)[source]¶

Insert a new event into Event_Log and return its auto-generated event_id.

Parameters:

save_id (str)
turn_id (int)
event_type (str)
target_entity (str)
payload (dict[str, Any])

Return type:

int

append_events_batch(events)[source]¶

Insert multiple events in a single transaction.

Parameters:: events (list[tuple[str, int, str, str, dict[str, Any]]]) – List of (save_id, turn_id, event_type, target_entity, payload) tuples.
Return type:: None

get_events(save_id, start_turn_id=0, up_to_turn_id=None)[source]¶

Fetch events for a save, ordered chronologically.

Parameters:

save_id (str) – The save whose events are requested.
start_turn_id (int) – Only events with turn_id > this value are returned.
up_to_turn_id (int | None) – If provided, only events with turn_id <= this value are returned. None means all future events.

Return type:

list[dict[str, Any]]

rebuild_state_cache(save_id, up_to_turn_id=None, force_full=False)[source]¶

Flush and rebuild State_Cache for a save by replaying Event_Log.

Parameters:

save_id (str) – The save whose cache is being rebuilt.
up_to_turn_id (int | None) – If provided, only events up to this turn are replayed. None means the full history.
force_full (bool) – If True, ignores all snapshots and replays the entire history from turn 0. Use this to fix cache corruption.

Return type:

None

Optimized to start from the nearest previous snapshot if one exists, unless force_full=True.

update_state_cache(save_id, events)[source]¶

Incrementally apply a just-appended batch of events to State_Cache.

State_Cache is a materialised view of the base stats derived from Event_Log. rebuild_state_cache() replays the entire history (or from the nearest snapshot); this method instead applies only the given batch on top of the affected entities’ current cached values, keeping the cache fresh after each turn without an O(history) replay.

This is what keeps DB reads (the sidebar’s load_full_game_state / load_stats tasks, which read State_Cache) in sync with the changes a turn just produced — see TICKET-002.

Parameters:

save_id (str) – The save whose cache is being updated.
events (list[tuple[str, int, str, str, dict[str, Any]]]) – List of (save_id, turn_id, event_type, target_entity, payload) tuples, in the same shape as append_events_batch(). Only entity_create / stat_change / stat_set events mutate the cache; all others are ignored.

Return type:

None

validate_integrity(save_id)[source]¶

Verify that the current State_Cache matches a fresh replay of history.

This is a diagnostic tool to detect corruption in the materialised cache. It does NOT use snapshots.

Returns:: A (passed, mismatches) tuple — mismatches maps entity_id to a dict of stat_key to (cached_val, actual_val) pairs.
Parameters:: save_id (str)
Return type:: tuple[bool, dict[str, Any]]

state_at(save_id, up_to_turn_id=None)[source]¶

Compute the materialised state by replaying events, without touching the DB.

Pure read: replays Event_Log (optionally up to a turn) and returns the resulting entity_id -> {stat_key: stat_value} map. Used by snapshotting and by the save editor (axiom.saves) to materialise state at any point.

Parameters:

save_id (str)
up_to_turn_id (int | None)

Return type:

dict[str, dict[str, str]]

take_snapshot(save_id, turn_id)[source]¶

Capture the current materialised state and store it in Snapshots.

This is an expensive operation (JSON-serialising the full state) and should be called sparingly.

Parameters:

save_id (str)
turn_id (int)

Return type:

None

get_current_stats(save_id, entity_id)[source]¶

Read the current materialised stats for one entity from State_Cache.

Parameters:

save_id (str) – The active save identifier.
entity_id (str) – The entity whose stats are requested.

Returns:

Dict mapping stat_key -> stat_value strings. Empty dict if the entity has no cached stats or does not exist.

Raises:

sqlite3.Error – On any database failure.

Return type:

dict[str, str]

Saves and event sourcing¶

axiom.savestore¶

axiom.saves¶

axiom.checkpoint¶

axiom.events¶

`axiom.savestore`¶

`axiom.saves`¶

`axiom.checkpoint`¶

`axiom.events`¶