System Design Document: Spatial-Temporal Campaign Chronicle Loop

J. Rogers, SE Ohio 

Target Architecture: The Codex Engine (Mega-Mappers Python Framework)

---

Executive Introduction

Traditional Virtual Tabletops (VTTs) operate as static calculators; they display map assets, track hit points, and hold text logs, but the software has no semantic understanding of the narrative happening within it. If a player kills an NPC, the software updates a number from 10 to 0, leaving the long-term storytelling consequences entirely up to the human Game Master's memory.

This blueprint modifies The Codex Engine into a Self-Sustaining Simulation Matrix.

By leveraging the engine's existing top-down, multi-layered data inheritance 

    Genre --> World --> Hex --> Room,

we are building a two-way event loop. As players sit around a physical table, an asynchronous recording thread captures their live voices, separates the speakers, and transcribes the speech. Simultaneously, the Pygame graphics and state controllers log every movement, wound, kill, and flight event.

Instead of generating flat text files, the engine routes these data points into a unified, time-stamped telemetry stream file. When a game session ends, a post-session local AI pipeline (via Ollama) evaluates this stream. Because every event is tied to an exact geographic coordinate and inherits a strict universal genre, the AI can reliably calculate complex, unscripted macro consequences—such as faction wars, generational blood feuds, or shifting town alignments—and write them directly back into the SQLite world database.

The following plan outlines the exact changes required to implement this architecture, explaining what we are changing, why we are changing it, and how it enables the final systemic goal.

---

Step 1: Database Infrastructure Upgrades (codex.db)

What we are doing

We are injecting two new relational tables into the core SQLite schema: entity_history and voice_registry.

Why we are doing it this way

Using a centralized SQL database provides a structured memory cache that local AI engines can query using Retrieval-Augmented Generation (RAG). By breaking the world history table down into strict spatial columns (such as local_hex_id and room_id), we completely avoid the "context window explosion" common in LLM architectures. The AI never has to read a massive, 100-page text chronicle of the entire campaign; instead, it is fed only the tiny, highly relevant historical anomalies associated with the exact coordinate pixel the players are currently standing on.

What this enables

• Geographic Event Isolation: Allows the code to filter and package history based entirely on where an action occurred.
• Persistent Vocal Identity: Maps abstract speaker signatures captured by microphone arrays back to concrete database entities (characters, monsters, or the DM) across multiple game sessions.

sql

-- Tracks historical events tied to specific coordinates across the top-down hierarchy
CREATE TABLE entity_history (
    history_id INTEGER PRIMARY KEY AUTOINCREMENT,
    session_id INTEGER,
    genre_key TEXT,            -- Inherited from the apex tier
    world_map_pixel TEXT,      -- Macro-scale location
    local_hex_id TEXT,         -- Regional-scale location
    dungeon_id TEXT,           -- Tactical-scale parent
    room_id TEXT,              -- Specific coordinate matrix
    timestamp_offset REAL,
    actor_name TEXT,           -- e.g., "Grog the Barbarian", "Goblin Chief"
    action_type TEXT,          -- "KILLED", "WOUNDED", "FLED", "SPOKE"
    narrative_summary TEXT     -- Cleaned, semantic summary for Ollama RAG injection
);

-- Active character voice maps for the local Diarization pipeline
CREATE TABLE voice_registry (
    speaker_tag TEXT PRIMARY KEY, -- e.g., "SPEAKER_01"
    character_name TEXT,          -- e.g., "Grog the Barbarian"
    is_dm INTEGER DEFAULT 0
);

This is the wrong design, the code does not use tables. It has nodes.

Step 2: The Asynchronous Telemetry Broker (chronicle.py)

What we are doing

We are implementing a thread-safe, memory-queued background broker class designed to capture live system events and write them sequentially to a line-delimited JSON (.jsonl) stream file.

Why we are doing it this way

In a real-time, ray-traced Pygame application running a dual-monitor multiprocessing environment, hard-drive write operations (Disk I/O) are an absolute bottleneck. If the main game loop halts to write text to a file every time an explosion happens or a character speaks, the frame rate will crater, ruining the tactical experience on the 32" monitor. By utilizing Python's queue.Queue inside a dedicated, isolated background thread, the game loop passes data into RAM instantly and moves on, leaving the background thread to handle streaming to the disk asynchronously.

What this enables

• Zero Frame-Rate Impact: Decouples game logic, asset rendering, and pathfinding calculations from file storage.
• Crash-Resistant Logging: Because data is continuously flushed to the .jsonl black box file on the fly, a sudden power loss or application crash will not corrupt the history file.

python

import json
import time
import queue
import threading

class CampaignChronicleBroker:
    def __init__(self, log_filename="session_telemetry.jsonl"):
        self.log_file = open(log_filename, "a", encoding="utf-8")
        self.event_queue = queue.Queue()
        self.start_time = time.time()
        
        # Dedicated background worker to keep disk I/O off the Pygame frame loop
        self.writer_thread = threading.Thread(target=self._drain_queue, daemon=True)
        self.writer_thread.start()

    def record_event(self, event_type, context_hierarchy, detail_dict):
        """
        Pushes a new structural log snapshot safely into the background processing queue.
        """
        payload = {
            "timestamp": round(time.time() - self.start_time, 2),
            "event_type": event_type,
            "hierarchy": context_hierarchy,
            "details": detail_dict
        }
        self.event_queue.put(payload)

    def _drain_queue(self):
        while True:
            event = self.event_queue.get()
            if event is None:
                break
            self.log_file.write(json.dumps(event) + "\n")
            self.log_file.flush()  # Force write to disk immediately without buffering lag
            self.event_queue.task_done()

    def close(self):
        self.event_queue.put(None)
        self.writer_thread.join()
        self.log_file.close()

Step 3: Local Voice Tracking & Diarization Engine (audio.py)

What we are doing

We are developing an isolated audio listener thread that continually samples the table environment, separates speech signals into discrete vocal tracks, and converts them to text via local Whisper bindings.

Why we are doing it this way

To maintain complete network privacy and enable offline play (the "cabin in the woods" factor), the audio extraction must occur locally without passing speech over public API vectors. The system runs local voice diarization (sorting signals into SPEAKER_00, SPEAKER_01) followed by transcription. Crucially, this script pulls a callback from the main engine coordinates the exact instant a word is spoken, appending current spatial metadata to the vocal string.

What this enables

• Hands-Free DM Operation: The DM does not need to pause play to type out summaries of conversations or negotiation outcomes; the engine records the verbatim dialogue natively.
• Dynamic Voice-to-Entity Binding: DMs can assign real-world player voices directly to character sheets inside the console interface, automatically turning generic speaker markers into narrative character actions.

python

import sounddevice as sd
import numpy as np
import scipy.io.wavfile as wav
import threading

class LocalVoiceTracker:
    def __init__(self, broker, context_fetcher_callback):
        self.broker = broker
        self.get_context = context_fetcher_callback
        self.is_recording = False
        self.voice_map = {} 

    def update_voice_mapping(self, speaker_tag, character_name):
        """Assigned dynamically by the DM in the UI console"""
        self.voice_map[speaker_tag] = character_name

    def audio_capture_loop(self):
        # Continually intercepts input audio chunks, separates speakers locally via 
        # voice embeddings, transcribes speech, and immediately fires a spatial telemetry event.
        while self.is_recording:
            audio_chunk, speaker_tag, raw_text = self._run_local_whisper_diarization()
            
            actor = self.voice_map.get(speaker_tag, speaker_tag)
            context = self.get_context() # Intercepts coordinate metrics exactly when spoken
            
            self.broker.record_event(
                event_type="transcript",
                context_hierarchy=context,
                detail_dict={"speaker_id": speaker_tag, "actor": actor, "text": raw_text}
            )
            
    def _run_local_whisper_diarization(self):
        # Hardware integration hook for local Whisper.cpp or WhisperX binary bindings
        pass

Step 4: State Mutation Injection (combat.py)

What we are doing

We are embedding structural event logging hooks directly inside the active engine controllers that modify creature variables (health, movement, status conditions).

Why we are doing it this way

Capturing dialogue alone is not enough to accurately simulate the reactive updates of a fantasy world. If a player screams "Die, monster!" and strikes a blow, an LLM reading only the transcript cannot know if the monster survived, fled down a dark tunnel, or fell in battle. By tying the logging loop into the mechanical state mutations of the game engine itself, we merge physical math (HP changes, grid coordinates) with semantic speech.

What this enables

• Action Context Synchronization: Guarantees that the transcript narrative perfectly matches the mechanical reality of the map board.
• Flawless Tracking of Escaped Threat Metrics: Explicitly registers when an NPC changes status to "FLED", signaling to the macro world processor that this specific entity is alive and can spread information to rival factions.

python

# Inserted into active codex_engine/controllers system classes
def modify_npc_hp(self, npc_id, damage_amount, engine_state):
    target_npc = self.get_npc(npc_id)
    target_npc.hp -= damage_amount
    
    # Extract complete, multi-tiered contextual inheritance data
    current_context = {
        "genre": engine_state.genre_blueprint.id,
        "world_pixel": engine_state.current_world_coords,
        "local_hex": engine_state.current_hex_id,
        "dungeon": engine_state.current_dungeon_id,
        "room": engine_state.current_room_id
    }
    
    # Evaluate systemic consequence vectors
    action = "WOUNDED"
    if target_npc.hp <= 0:
        action = "KILLED"
    elif target_npc.is_fleeing:
        action = "FLED"

    # Stream event payload off-thread to maintain ray-tracing render integrity
    engine_state.chronicle_broker.record_event(
        event_type="npc_mutation",
        context_hierarchy=current_context,
        detail_dict={
            "npc_id": npc_id,
            "npc_name": target_npc.name,
            "action": action,
            "current_hp": target_npc.hp,
            "hp_delta": -damage_amount
        }
    )

Step 5: Local Ollama Post-Session Macro Evaluation Engine

What we are doing

We are developing a post-game execution script that parses the .jsonl stream log, organizes the session events by their geographical hierarchy IDs, passes them through a genre-bounded local AI prompt (via Ollama), and mutates the parent database rows.

Why we are doing it this way

This step closes the loop, transforming raw logs into active database parameters. Instead of using unconstrained prompting, we construct a strict context box. The local AI is provided with the overarching genre rules, the baseline environment, and the matching physical coordinates. We then force the AI to return data in a strict JSON schema. This completely eliminates model hallucination, ensuring that a small, fast local model (like Llama 3 or Mistral) maintains absolute narrative and thematic consistency.

What this enables

• Unscripted Bottom-Up World Mutation: Changes the macro-world state automatically based on micro tactical actions. Killing a dungeon boss changes the random encounter lists across neighboring local hex coordinates, while slaying a merchant mutates the political text of an entire kingdom tier.
• Infinite, Consistent Campaign Continuity: Ensures that when the party returns to an area several sessions later, the updated maps, room descriptions, and ambient town lore dynamically reflect the permanent consequences of their past actions.

python

import json
import ollama

def process_session_consequences(jsonl_log_path, db_connection):
    # Sorts the raw stream log file, pairing chronological actions into distinct hex/world keys
    grouped_data = aggregate_log_by_location(jsonl_log_path)
    
    for local_hex, data in grouped_data.items():
        # Inject structural, hierarchical limits before writing text prompts
        prompt = f"""
        [GENRE INHERITANCE]: {data['genre']}
        [REGIONAL ENVIRONMENT]: {data['hex_description']}
        [DUNGEON FACTION ARCHETYPE]: {data['faction_state']}
        
        [LOGGED SESSION EVENTS]:
        {data['serialized_events_and_transcripts']}
        
        Analyze these events through the exact framework of the genre.
        Output a strict JSON object mapping structural changes to the environment:
        {{
            "hex_state_mutation": "How the overall regional map text description changes",
            "faction_changes": "How local power structures shift or split",
            "world_ripple_effects": "Concrete long-term consequences (e.g., vengeance oaths, shifting borders)"
        }}
        """
        
        # Dispatch structured query to the local offline edge model
        response = ollama.generate(model="mistral", prompt=prompt, format="json")
        
        # Parse the JSON response and execute UPDATE metrics directly back into the SQLite schema
        apply_database_mutations(local_hex, response, db_connection)

def aggregate_log_by_location(jsonl_log_path):
    # Aggregation parsing code logic goes here
    pass

def apply_database_mutations(local_hex, ai_json_response, db_connection):
    # SQLite update queries execution code goes here
    pass

Next Development Priorities

To begin coding these modules, choose an initial focus area:

• Setting up the local voice recording dependencies (WhisperX vs. direct pyannote bindings)
• Integrating the CampaignChronicleBroker queue class into your core Pygame initialization loop
• Building the JSON log aggregator algorithm that compiles the raw text strings for the local AI prompt