Runtime internals¶
The narration pipeline and its helpers. Most applications drive these through
axiom.session.Session rather than directly.
axiom.arbitrator¶
core/arbitrator.py
The ArbitratorEngine — Axiom AI’s deterministic firewall between LLM creativity and the game’s mathematical state.
On every narrative turn the ArbitratorEngine:
Fetches current entity stats from State_Cache + applies modifier overlay.
Retrieves relevant narrative memories from VectorMemory (RAG).
Builds the full narrative prompt (injecting any pending correction).
Calls the LLM and parses its response.
Validates every proposed state change against current stats.
Persists valid changes via EventSourcer; queues corrections for invalids.
Runs the Rules Engine for each mutated entity; persists triggered actions.
Ticks modifier durations.
Embeds the narrative chunk into VectorMemory.
Returns an ArbitratorResult with full detail for the UI / tests.
The Correction Loop (spec §4-B)¶
If a change is rejected, a hidden system message is stored in _pending_correction. On the VERY NEXT turn this message is injected into the prompt immediately before the user’s input, then immediately cleared so it cannot affect turn N+2.
- class axiom.arbitrator.ArbitratorResult(narrative_text, applied_changes=<factory>, rejected_changes=<factory>, inventory_changes=<factory>, triggered_rules=<factory>, rule_chain_warning=False, game_state_tag='exploration', player_entity_id='player', elapsed_minutes=1, scene_pace='deliberate', image_path=None, in_game_time=0)[source]¶
The complete output of one ArbitratorEngine turn.
- Parameters:
- applied_changes¶
State changes that passed validation and were persisted.
- rejected_changes¶
State changes that failed validation, each augmented with a “reason” key explaining the failure.
- rule_chain_warning¶
True if the rules engine reached its iteration limit, indicating a possible infinite loop in creator rules.
- Type:
- class axiom.arbitrator.ArbitratorEngine(db_path, rules_list)[source]¶
Validates and applies LLM-proposed state changes for one narrative turn.
- Parameters:
- configure(llm, vector_memory, time_llm=None)[source]¶
Inject runtime dependencies before process_turn.
- Parameters:
llm (LLMBackend)
vector_memory (VectorMemory)
time_llm (LLMBackend | None)
- Return type:
None
- invalidate_stats_cache()[source]¶
No-op kept for API compatibility.
Effective stats (base + modifier overlay) are now re-read from State_Cache + Active_Modifiers on every turn (see _fetch_effective_stats), so there is no stale cross-turn cache to clear. Previously this dropped an in-memory snapshot that could keep an expired modifier’s delta baked in until the next chronicler/rewind — callers (rewind, post-chronicler) still call this harmlessly.
- Return type:
None
- process_turn(save_id, turn_id, intents, universe_system_prompt, history, stream_token_callback=None, temperature=0.7, top_p=1.0, verbosity_level='balanced', mode='Normal', hero_entity_id=None)[source]¶
Execute one full ArbitratorEngine turn and return the result.
- Parameters:
save_id (str) – The active save identifier.
turn_id (int) – The current turn number (monotonically increasing).
intents (dict[str, str]) – Dict mapping actor entity_id to their intent text.
universe_system_prompt (str) – The universe’s foundational system prompt.
history (list[LLMMessage]) – Prior conversation turns for context.
stream_token_callback (Callable[[str], None] | None) – Optional callable invoked with each streaming token as it arrives from the LLM.
temperature (float) – Sampling temperature (0.0 to 1.0).
top_p (float) – Nucleus sampling parameter (0.0 to 1.0).
verbosity_level (str) – ‘short’, ‘balanced’, or ‘talkative’.
mode (str) – Game mode (‘Normal’, ‘Hardcore’, ‘Companion’).
hero_entity_id (str | None) – Optional ID of the AI Hero entity.
- Returns:
ArbitratorResult containing narrative text and all change outcomes.
- Raises:
LLMConnectionError – If the LLM backend is unreachable.
- Return type:
axiom.chronicler¶
core/chronicler.py
The Chronicler Engine — Axiom AI’s macro-simulation agent.
The Chronicler simulates the independent evolution of the off-screen world (factions, VIP NPCs, cities) without the player’s involvement. It runs periodically (every N player turns, or on explicit time-skip events) and produces JSON tool calls that update entity stats in the database.
Design principles¶
The Chronicler NEVER raises on malformed LLM responses. A broken world simulation is always preferable to a crashed game session.
All updates are written to Event_Log with event_type=’chronicler_update’ so they are included in checkpoint replays and can be rewound.
World_Tension_Level (stored in Universe_Meta) throttles the severity of simulated events: low tension → mundane shifts, high tension → cataclysms.
- class axiom.chronicler.ChroniclerResult(updated_entities=<factory>, events_appended=0, world_tension_used=0.3, world_news=<factory>)[source]¶
The complete output of one Chronicler world-simulation run.
- Parameters:
- class axiom.chronicler.ChroniclerEngine(llm, event_sourcer, db_path, trigger_interval=720)[source]¶
Simulates the off-screen world by calling the LLM periodically.
- Parameters:
llm (LLMBackend) – The LLM backend used for world simulation calls.
event_sourcer (EventSourcer) – Handles Event_Log writes and State_Cache reads.
db_path (str) – Path to the universe .db for entity and meta queries.
trigger_interval (int) – In-game minutes between automatic Chronicler runs. Defaults to 720 (12 in-game hours).
- should_trigger(current_time, previous_time)[source]¶
Return True if the in-game clock crossed a trigger_interval boundary.
The Chronicler is driven by in-game minutes, not player turns: it fires once whenever the world clock moves from one trigger_interval-minute block into a later one. This makes a single long time-skip trigger exactly one off-screen simulation, while many short turns accumulate until they cross the next boundary (Pilier 5 / TICKET-018).
Pure function — no I/O.
- Parameters:
- Returns:
True when previous_time and current_time fall in different trigger_interval-minute blocks (i.e. a boundary was crossed).
- Return type:
- run(save_id, turn_id, temperature=0.7, top_p=1.0)[source]¶
Execute one Chronicler world-simulation cycle.
- Steps:
Read World_Tension_Level from Universe_Meta.
Fetch all active off-screen entities and their current stats.
Build the Chronicler prompt.
Call the LLM (non-streaming; expects only a JSON tool call).
Parse the resulting state_changes.
Validate each change (entity must exist and be active).
Persist valid changes via EventSourcer.
Return ChroniclerResult.
On any malformed LLM response the method logs a warning internally and returns an empty ChroniclerResult — it never raises.
- Parameters:
- Returns:
ChroniclerResult summarising what changed.
- Return type:
- force_trigger(save_id, turn_id)[source]¶
Explicitly trigger a Chronicler run regardless of turn threshold.
Intended for time-skip actions (sleeping, fast travel) where the player has explicitly advanced time. Semantically equivalent to run() but named distinctly for call-site clarity.
- Parameters:
- Returns:
ChroniclerResult from the simulation run.
- Return type:
axiom.memory¶
llm_engine/vector_memory.py
Local vector-database memory for Axiom AI narrative chunks.
Every piece of narrative embedded here carries a turn_id metadata tag. This enables the surgical rollback required by the Checkpoint system: when the player rewinds to turn N, all chunks with turn_id > N are permanently deleted so they cannot bleed into the rebuilt timeline.
Backend: ChromaDB (persistent, local) Embedding model: sentence-transformers all-MiniLM-L6-v2 (fully offline)
Collection layout¶
Collection name : “narrative_memory” Document : the text chunk Metadata fields : save_id (str), turn_id (int), chunk_type (str) ID : UUID string, generated per chunk
- axiom.memory.preload_embedding_runtime()[source]¶
Force torch’s native runtime to load on the calling (main) thread.
The sentence-transformers embedding model is loaded and used on worker threads (VectorInitWorker / NarrativeWorker). The first encode lazily pulls in
torch._dynamo→triton, whichdlopen()``s ``libtriton.so. Doing thatdlopenfrom a secondary thread while Qt is running segfaults (native crash, no Python traceback). Importing it once here, on the main thread at startup, makes the later cross-thread use safe.Call this from the GUI/CLI entry point before any worker thread touches VectorMemory. Idempotent, never raises. Returns True if the runtime was pre-loaded, False if torch is unavailable (e.g. headless test stubs).
- Return type:
- class axiom.memory.VectorMemory(persist_dir)[source]¶
Local semantic memory store backed by ChromaDB.
- Parameters:
persist_dir (str) – Filesystem path where ChromaDB will store its data. Created automatically if it does not exist.
- embed_chunk(save_id, turn_id, text, chunk_type='narrative')[source]¶
Embed a text chunk and store it with turn_id metadata.
- query(save_id, query_text, k=5, current_turn_id=None, max_turn_id=None)[source]¶
Retrieve the top-k most relevant chunks using Time-Weighted search.
axiom.time_system¶
core/time_system.py
Flexible time and calendar system for Axiom AI. Allows custom minutes per hour, hours per day, and named months.
- class axiom.time_system.TimeComponents(year, month_name, day, hour, minute, phase_key)[source]¶
Decomposition of an instant into raw data (zero presentation/translation).
phase_key is a stable key among dawn/morning/afternoon/dusk/night; month_name comes from the universe calendar (data, not a translation).
- class axiom.time_system.CalendarConfig(minutes_per_hour=60, hours_per_day=24, days_per_month=<factory>, month_names=<factory>, start_day=1, start_hour=0, start_minute=0)[source]¶
Configuration for a custom calendar.
- class axiom.time_system.TimeSystem(config=None)[source]¶
Handles time conversion and formatting based on a CalendarConfig.
- Parameters:
config (CalendarConfig | None)
- get_time_components(total_minutes)[source]¶
Decompose cumulative minutes into (year, month, day, h, min, phase key).
Raw data only: no translation. The frontend localises the display from these fields.
- Parameters:
total_minutes (int)
- Return type:
- get_time_string(total_minutes)[source]¶
Default English rendering (dev / CLI / library). Zero engine-side localisation.
The GUI does NOT go through here: it formats through its own localisation layer to display in the user’s language.
axiom.rules¶
core/rules_engine.py
JSON-based Rules Engine for Axiom AI.
The engine evaluates a set of creator-defined rules against an entity’s current stats and returns the list of actions that should be applied.
Canonical Rule JSON schema:
{
"rule_id": "str",
"priority": int, // lower number = higher priority
"target_entity": "str | '*'", // '*' means applies to any entity
"conditions": {
"operator": "AND" | "OR",
"clauses": [
{
"stat": "str",
"comparator": "<=" | ">=" | "==" | "!=" | "<" | ">",
"value": number | "str"
},
// nested condition groups are also supported:
{
"operator": "AND" | "OR",
"clauses": [ ... ]
}
]
},
"actions": [
{
"type": "stat_change" | "stat_set" | "trigger_event" | "set_status",
"target": "str", // entity_id to affect
"stat": "str", // stat key (for stat_change / stat_set)
"delta": number, // signed delta (for stat_change)
"value": "str" | number // absolute value (for stat_set / set_status)
}
]
}
Notes
Rules are evaluated in ascending priority order (0 = highest priority).
A rule with target_entity == ‘*’ is evaluated for every entity.
A rule with a specific target_entity is only evaluated when entity_id matches.
apply_actions() is a pure function: it does NOT write to the database. The caller (Arbitrator, Phase 2) is responsible for persisting changes via EventSourcer.
- class axiom.rules.RulesEngine(rules)[source]¶
Evaluates JSON rules against entity stats and produces triggered actions.
- Parameters:
rules (list[dict[str, Any]]) – List of rule dicts loaded from the Rules table of a universe db. Rules are sorted by priority at construction time.
- evaluate(entity_id, stats)[source]¶
Evaluate all rules against an entity’s current stats.
- Rules are processed in priority order. A rule fires when:
Its target_entity is ‘*’ OR equals entity_id, AND
Its conditions evaluate to True.
- Parameters:
- Returns:
List of action dicts from all triggered rules, in priority order. May be empty if no rules fire.
- Return type:
- apply_actions(actions, stats)[source]¶
Apply a list of actions to a stats snapshot and return the updated copy.
This is a pure function. The original stats dict is not mutated.
Supported action types:
stat_change — adds ‘delta’ (float) to an existing or zero-valued stat.
stat_set — unconditionally sets a stat to the string form of ‘value’.
set_status — alias for stat_set; sets ‘stat’ to string ‘value’.
trigger_event — no immediate stat mutation; included in output for the caller to dispatch as a new Event_Log entry.
axiom.modifiers¶
database/modifier_processor.py
Active Modifiers management for Axiom AI.
Modifiers are temporary stat adjustments (buffs, debuffs, curses, blessings) that automatically expire after a set amount of in-game time (minutes). This module provides:
apply_modifiers: overlay active modifiers on top of a base stat snapshot for real-time display (read-only, no DB mutation).
tick_modifiers: decrement all modifier durations by elapsed minutes and purge expired ones from the database.
add_modifier: insert a new modifier row and return its ID.
Modifiers live in the Active_Modifiers table, which is keyed on (modifier_id, entity_id, stat_key). Every modifier carries a signed float delta and a minutes_remaining countdown. When minutes_remaining reaches 0 the modifier is considered expired and will be deleted on the next tick.
- class axiom.modifiers.ModifierProcessor(db_path)[source]¶
Manages temporary stat modifiers for entities in a universe database.
- Parameters:
db_path (str) – Filesystem path to an existing universe .db file created by database.schema.create_universe_db().
- apply_modifiers(save_id, entity_id, base_stats)[source]¶
Overlay active modifiers on a base stat snapshot.
Reads all active modifiers for the entity and adds their deltas to the corresponding numeric stats. Non-numeric stats are left unchanged. This method is read-only: it does not write to the database.
- Parameters:
save_id (str) – The active save (used to scope modifier lookup via entity_id — modifiers are stored per entity, not per save, but the entity must belong to this save’s universe).
entity_id (str) – The entity whose modifiers are applied.
base_stats (dict[str, str]) – The entity’s current base stat snapshot from State_Cache (stat_key -> string value).
- Returns:
New dict with modifier deltas applied. Keys not affected by any modifier are copied verbatim from base_stats.
- Raises:
sqlite3.Error – On any database failure.
- Return type:
- tick_modifiers(save_id, elapsed_minutes=1)[source]¶
Decrement minutes_remaining for all modifiers of this save’s entities.
- For every active modifier associated with any entity in the database:
Decrements minutes_remaining by elapsed_minutes.
Deletes the modifier if minutes_remaining reaches 0 after decrement.
The save_id parameter scopes the tick to entities relevant to the given save by joining through the State_Cache (entities that have cache entries for this save). If no scoping information exists, all entity modifiers in Active_Modifiers are ticked.
- Parameters:
- Returns:
List of modifier_id strings that were deleted (expired) this tick.
- Raises:
sqlite3.Error – On any database failure.
- Return type:
- add_modifier(save_id, entity_id, stat_key, delta, minutes)[source]¶
Insert a new modifier and return its generated modifier_id.
- Parameters:
save_id (str) – The save this modifier is associated with.
entity_id (str) – The entity the modifier affects.
stat_key (str) – The stat key the delta is applied to.
delta (float) – Signed float adjustment (positive = buff, negative = debuff).
minutes (int) – Number of in-game minutes the modifier lasts (must be >= 1).
- Returns:
The newly generated modifier_id (UUID string).
- Raises:
ValueError – If minutes < 1.
sqlite3.Error – On any database failure.
- Return type:
axiom.multiplayer¶
axiom.multiplayer — sequential turn-resolution queue.
In multiplayer, player actions are resolved one at a time (FIFO) to avoid any race on the database. Pure threading, zero Qt — the Qt shell (core/multiplayer_queue.py::ArbitratorWorker) merely moves run_loop onto a QThread and translates the callbacks into signals.
- class axiom.multiplayer.PlayerAction(player_id, text, save_id, turn_id, universe_system_prompt, history, temperature=0.7, top_p=1.0, verbosity_level='balanced')[source]¶
A player action awaiting resolution.
- class axiom.multiplayer.ActionQueue(arbitrator)[source]¶
FIFO queue of player actions, resolved sequentially by the arbitrator.
run_loop is blocking: the caller runs it on ITS thread (QThread on the GUI side, plain Python thread headless). enqueue and stop are thread-safe and callable from anywhere.
- Parameters:
arbitrator (ArbitratorEngine)
- enqueue(action)[source]¶
Add an action to resolve.
- Parameters:
action (PlayerAction)
- Return type:
None
axiom.db_helpers¶
workers/db_helpers.py
Synchronous database helper functions for one-time UI bootstrap reads.
These are small, fast, lightweight operations that are acceptable to run on the main thread during view construction or session initialisation (e.g. reading 1–2 rows of metadata at session start).
All SQL strings in the project are concentrated in database/ and workers/ modules — never in ui/ files — to satisfy the MVC separation mandate.
- axiom.db_helpers.apply_stat_preset(db_path, preset_name)[source]¶
Apply a stat preset to a universe database.
- axiom.db_helpers.read_universe_card_metadata(db_path)[source]¶
Read display metadata for a universe card widget.
- axiom.db_helpers.provision_blank_universe(db_path, name)[source]¶
Insert default Universe_Meta rows into a freshly provisioned database.
- axiom.db_helpers.create_new_save(db_path, player_name, difficulty, player_persona='')[source]¶
Insert a new save row and return its save_id.
- axiom.db_helpers.load_saves(db_path)[source]¶
Read all saves for a universe, sorted most-recent first.
Runs the player_persona migration automatically so older databases remain compatible.
- Parameters:
db_path (str) – Path to the universe .db file.
- Returns:
save_id, player_name, difficulty, last_updated, player_persona.
- Return type:
List of save dicts with keys
- axiom.db_helpers.load_rules_for_session(db_path)[source]¶
Read all rules from a universe database for session initialisation.
- axiom.db_helpers.load_active_entities(db_path)[source]¶
Read all active entities (with their stats) for session initialisation.
Mirrors the shape produced by workers/db_worker.load_entities_and_rules (the list the GUI feeds to NarrativeWorker), so a headless Session can resolve the Companion Hero entity itself instead of relying on the UI.
- Parameters:
db_path (str) – Path to the universe .db file.
- Returns:
{entity_id, entity_type, name, description, stats}. Empty list if the database cannot be read.
- Return type:
List of entity dicts
- axiom.db_helpers.get_max_turn_id(db_path, save_id)[source]¶
Read the highest turn_id from Event_Log for a save (session resume).
- axiom.db_helpers.get_current_time(db_path, save_id)[source]¶
Read the highest in_game_time from Timeline for a save.
- axiom.db_helpers.get_time_of_day_context(total_minutes)[source]¶
Convert total minutes into a descriptive Day, Time, and Phase string.
- Parameters:
total_minutes (int) – Cumulative in-game minutes.
- Returns:
“Day X, HH:MM (Phase)”
- Return type:
Formatted string
- axiom.db_helpers.get_inventory(db_path, save_id, entity_id)[source]¶
Fetch the inventory for a specific entity in a save.
axiom.textfmt¶
axiom.textfmt — language-neutral text formatting, engine-side.
fmt_num is NOT translation: it is display cleanup for numbers (avoids “3.0” or “0.1000000001”). The engine needs it regardless of any language; localisation itself lives on the frontend side (see core.localization).