mud_server.ledger.writer ======================== .. py:module:: mud_server.ledger.writer .. autoapi-nested-parse:: 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 (:func:`append_event` and :func:`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/.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: .. code-block:: text { "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": { "" }, "_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 ----------------- :exc:`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 ---------- .. autoapisummary:: mud_server.ledger.writer.logger Exceptions ---------- .. autoapisummary:: mud_server.ledger.writer.LedgerWriteError Classes ------- .. autoapisummary:: mud_server.ledger.writer.LedgerVerifyResult Functions --------- .. autoapisummary:: mud_server.ledger.writer.append_event mud_server.ledger.writer.verify_world_ledger Module Contents --------------- .. py:data:: logger .. py:exception:: LedgerWriteError Bases: :py:obj:`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. .. attribute:: 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. .. py:class:: LedgerVerifyResult Result of a ledger integrity check performed by :func:`verify_world_ledger`. .. attribute:: 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. .. attribute:: last_event_id The ``event_id`` of the last valid event, or ``None`` if the ledger is empty or corrupt. .. attribute:: error_detail Human-readable description of the failure reason, or ``None`` if status is ``"ok"`` or ``"empty"``. .. py:attribute:: status :type: Literal['ok', 'empty', 'corrupt'] .. py:attribute:: last_event_id :type: str | None .. py:attribute:: error_detail :type: str | None .. py:function:: append_event(world_id, event_type, data, *, ipc_hash = None, meta = None) 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. :param world_id: World this event belongs to. Must be non-empty. Used as the filename stem: ``data/ledger/.jsonl``. :param event_type: Dot-namespaced event type, e.g. ``"chat.translation"`` or ``"chat.mechanical_resolution"``. Must be non-empty. :param data: Event-specific payload dict. Must be JSON-serialisable. The schema is defined by the event type; this module is opaque to the payload content. :param ipc_hash: Optional IPC hash produced by the axis engine. ``None`` is valid and honest for pre-axis-engine events; it is stored as JSON ``null``. :param meta: 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. :raises LedgerWriteError: If the filesystem write fails for any reason (disk full, permission denied, invalid path). Callers must catch this and log a warning. 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) .. py:function:: verify_world_ledger(world_id) 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. :param world_id: The world whose ledger to verify. :returns: A :class:`LedgerVerifyResult` describing the outcome. The ``last_event_id`` field is populated on ``"ok"`` status and is ``None`` for ``"empty"`` or ``"corrupt"``. 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)