mud_server.ledger.writer

JSONL ledger writer for the PipeWorks axis engine.

Overview

This module is the single implementation file for the ledger package. It exposes two public functions (append_event() and verify_world_ledger()) and the types they return.

The ledger is the authoritative record of all axis mutations. The SQLite database is a materialized view derived from the ledger. The sequence is always:

  1. Compute axis deltas (axis engine).

  2. Write a chat.mechanical_resolution event to the JSONL ledger. ← authoritative

  3. Apply deltas to the DB (materialization step). ← derived

Until the axis engine is built, chat.translation events are written with ipc_hash: null and meta.phase = "pre_axis_engine". These are honest incomplete records, not errors.

Storage

Each world’s events are stored in a single JSONL file:

data/ledger/<world_id>.jsonl

One file per world. Events are appended in timestamp order (append order == wall-clock order in a single-process server). The directory and file are created automatically on the first write.

Envelope format

Every line is a self-contained JSON object with the following fields:

{
  "event_id":       "a3f91c9e2d4b5e6f...",
  "timestamp":      "2026-02-27T14:23:01.452345+00:00",
  "world_id":       "daily_undertaking",
  "event_type":     "chat.translation",
  "schema_version": "1.0",
  "ipc_hash":       null,
  "meta":           {"phase": "pre_axis_engine"},
  "data":           { "<event-specific payload>" },
  "_checksum":      "sha256:b94f3e..."
}

_checksum is computed over the JSON-serialized envelope body (all fields except _checksum itself, serialized with sort_keys=True). This allows corruption detection without the checksum being part of its own input.

Event type namespace

chat.mechanical_resolution    axis engine: chat resolver
chat.translation              translation layer
environment.condition_applied axis engine: environmental resolver (future)
action.physical_resolved      axis engine: physical resolver (future)
outcome.economic_resolved     axis engine: economic resolver (future)
axis.manual_override          admin/superuser override (future)

Concurrency

fcntl.flock(LOCK_EX) is acquired before every append and released after flush(). This serialises concurrent writers within a single process and across multiple processes on the same host.

Platform note: fcntl is POSIX-only (Darwin + Linux). Windows is not supported by the ledger writer. If Windows support is needed, replace fcntl.flock with a cross-platform locking library such as filelock.

Failure isolation

LedgerWriteError is raised on filesystem failure. Callers must catch it, log a warning, and continue. A ledger failure is never fatal to the calling game interaction — only the audit record is lost.

# Correct caller pattern:
try:
    append_event(...)
except LedgerWriteError:
    logger.warning("Ledger write failed — interaction continues.", exc_info=True)

TODO(ledger-hardening): In a production deployment, a failed ledger write should trigger an alert and degrade the affected world to read-only mode. The PoC treats ledger failures as non-fatal warnings.

Attributes

logger

Exceptions

LedgerWriteError

Raised when a ledger append fails due to a filesystem or encoding error.

Classes

LedgerVerifyResult

Result of a ledger integrity check performed by verify_world_ledger().

Functions

append_event(world_id, event_type, data, *[, ...])

Append one event to the world's JSONL ledger file.

verify_world_ledger(world_id)

Verify the integrity of the most recent event in a world's ledger.

Module Contents

mud_server.ledger.writer.logger
exception mud_server.ledger.writer.LedgerWriteError[source]

Bases: Exception

Raised when a ledger append fails due to a filesystem or encoding error.

Callers must catch this exception, log a warning, and allow the game interaction to continue. A ledger write failure must never propagate to the player-facing response.

message

Human-readable description of the failure.

Example:

try:
    append_event(world_id, event_type, data)
except LedgerWriteError as exc:
    logger.warning("Ledger write failed: %s", exc)

Initialize self. See help(type(self)) for accurate signature.

class mud_server.ledger.writer.LedgerVerifyResult[source]

Result of a ledger integrity check performed by verify_world_ledger().

status

One of: - "ok" — last event is valid JSON and checksum matches. - "empty" — file does not exist or contains no events. - "corrupt" — last line is malformed JSON or checksum mismatch.

last_event_id

The event_id of the last valid event, or None if the ledger is empty or corrupt.

error_detail

Human-readable description of the failure reason, or None if status is "ok" or "empty".

status: Literal['ok', 'empty', 'corrupt']
last_event_id: str | None
error_detail: str | None
mud_server.ledger.writer.append_event(world_id, event_type, data, *, ipc_hash=None, meta=None)[source]

Append one event to the world’s JSONL ledger file.

This is the only authorised write path for all axis events. The caller must never write directly to the JSONL file.

The function:

  1. Validates world_id and event_type.

  2. Generates a unique event_id (UUID4 hex) and ISO-8601 UTC timestamp.

  3. Assembles the envelope dict (all fields except _checksum).

  4. Computes a SHA-256 checksum over the canonical JSON serialisation of the envelope body.

  5. Appends the completed envelope as a single newline-terminated JSON line under an exclusive POSIX file lock.

The ledger directory and file are created if they do not yet exist.

Parameters:

  • world_id (str) – World this event belongs to. Must be non-empty. Used as the filename stem: data/ledger/<world_id>.jsonl.

  • event_type (str) – Dot-namespaced event type, e.g. "chat.translation" or "chat.mechanical_resolution". Must be non-empty.

  • data (dict) – Event-specific payload dict. Must be JSON-serialisable. The schema is defined by the event type; this module is opaque to the payload content.

  • ipc_hash (str | None) – Optional IPC hash produced by the axis engine. None is valid and honest for pre-axis-engine events; it is stored as JSON null.

  • meta (dict | None) – Optional metadata dict for phase markers and diagnostic fields, e.g. {"phase": "pre_axis_engine"}. If None, an empty dict is stored.

Returns:

The event_id of the written event as a 32-character lowercase hex string (UUID4 without hyphens).

Raises:

  • ValueError – If world_id or event_type is empty or blank.

  • LedgerWriteError – If the filesystem write fails for any reason (disk full, permission denied, invalid path). Callers must catch this and log a warning.

Return type:

str

Example:

event_id = append_event(
    world_id="daily_undertaking",
    event_type="chat.translation",
    data={"status": "success", "ic_output": "The shadows suit you."},
    ipc_hash=None,
    meta={"phase": "pre_axis_engine"},
)
logger.debug("Ledger event written: %s", event_id)
mud_server.ledger.writer.verify_world_ledger(world_id)[source]

Verify the integrity of the most recent event in a world’s ledger.

Intended to be called at server startup to detect corruption before the first write. Only the last non-empty line is inspected; a full replay verification (reading every line from the beginning) is future scope.

The check performs two assertions:

  1. The last line deserialises as valid JSON.

  2. The _checksum field in that line matches the SHA-256 computed from the line body (all fields except _checksum, serialised with sort_keys=True).

If the last line is corrupt, the caller should log a CRITICAL-level warning. The server is not blocked from starting in the PoC — this is a diagnostic tool, not an enforced gate. Mark with TODO(ledger-hardening) when upgrading to a production guard.

Parameters:

world_id (str) – The world whose ledger to verify.

Returns:

A LedgerVerifyResult describing the outcome. The last_event_id field is populated on "ok" status and is None for "empty" or "corrupt".

Return type:

LedgerVerifyResult

Example:

result = verify_world_ledger("daily_undertaking")
if result.status == "corrupt":
    logger.critical(
        "Ledger integrity failure for world %r: %s",
        "daily_undertaking",
        result.error_detail,
    )
elif result.status == "ok":
    logger.info("Ledger OK, last event: %s", result.last_event_id)