orchestrata_ultimate.factory.enforcement.ledger

Reference › orchestrata_ultimate › factory › enforcement › ledger

ledger holds the two stores behind governed-write enforcement (SPEC-033 section 3, row B1c, Phase 1). Both use the locked, atomic file pattern of the assumption ledger: an exclusive fcntl.flock on a sidecar lock file around each read-modify-write, and whole-file rewrites through a temporary file and os.replace. _LockedFile provides both the JSONL and the single-JSON-document operations.

EventLedger is an append-only JSONL file with one GovernedWriteRecord per line. record appends a new record for a writer, its materials, products, worktree and commit; records_for returns the records for a worktree and commit pair; all returns everything. There is no update or rewrite, since a governed write is a historical fact. WriterRegistry is a single JSON file listing registered WriterIdentity values. register records an identity and raises ValueError if the by argument is empty (registration is a human act, per ADR-0009); registered tests whether a (binary hash, config hash) pair is registered; entries lists them. enforce and bootstrap read them, as do the red-first ledger and the promotion modules. Types come from models.

At a glance

Class diagram

Dependency diagram

API

Classes

_LockedFile

Line 83

Shared locked/atomic-write primitive for the ledger and the registry.

EventLedger

Line 148

Append-only JSONL ledger of governed-write events (B1c).

WriterRegistry

Line 208

JSON-file registry of governed-writer identities (B1c).

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/factory/enforcement/ledger.py

"""``EventLedger`` + ``WriterRegistry`` (SPEC-033 §3, B1c, Phase 1).

Both stores use atomic + locked I/O mirroring the pattern in
``orchestrata_ultimate.factory.assumptions.ledger``: an exclusive
``fcntl.flock`` on a sidecar lock file guards every read-modify-write, and
full rewrites go through a sibling temp file + ``os.replace`` (atomic on
POSIX) so a crash mid-write never leaves a torn file. Two threads (or
processes) appending concurrently serialize on the lock instead of
corrupting the file.

``EventLedger`` is append-only JSONL (one ``GovernedWriteRecord`` per line);
records are immutable once written — there is no ``update``/``rewrite``
method, unlike the assumption ledger, because a governed-write record is a
historical fact, not a lifecycle object. ``WriterRegistry`` is a single JSON
file (a list of registered-identity entries), since the whole set is small
and read as a unit on every ``registered()`` check.
"""

from __future__ import annotations

import fcntl
import json
import os
from datetime import UTC, datetime
from pathlib import Path
from typing import IO, Any

from orchestrata_ultimate.factory.enforcement.models import (
    GovernedWriteRecord,
    WriterIdentity,
)

# The lock-file handle type shared by _acquire/_release below.
_LockFile = IO[str]


def _utc_now_iso() -> str:
    return datetime.now(UTC).isoformat()


def _writer_to_dict(writer: WriterIdentity) -> dict[str, str]:
    return {
        "name": writer.name,
        "binary_hash": writer.binary_hash,
        "config_hash": writer.config_hash,
    }


def _writer_from_dict(data: dict[str, Any]) -> WriterIdentity:
    return WriterIdentity(
        name=data["name"],
        binary_hash=data["binary_hash"],
        config_hash=data["config_hash"],
    )


def _record_to_dict(record: GovernedWriteRecord) -> dict[str, Any]:
    return {
        "record_id": record.record_id,
        "writer": _writer_to_dict(record.writer),
        "materials": dict(record.materials),
        "products": dict(record.products),
        "worktree": record.worktree,
        "commit": record.commit,
        "byproducts": dict(record.byproducts),
        "recorded_at": record.recorded_at,
    }


def _dict_to_record(data: dict[str, Any]) -> GovernedWriteRecord:
    return GovernedWriteRecord(
        record_id=data["record_id"],
        writer=_writer_from_dict(data["writer"]),
        materials=dict(data.get("materials", {})),
        products=dict(data.get("products", {})),
        worktree=data["worktree"],
        commit=data["commit"],
        byproducts=dict(data.get("byproducts", {})),
        recorded_at=data["recorded_at"],
    )


class _LockedFile:
    """Shared locked/atomic-write primitive for the ledger and the registry."""

    def __init__(self, path: Path):
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.lock_path = self.path.with_name(self.path.name + ".lock")

    def _acquire(self) -> _LockFile:
        # 'a' (not 'w') so we never truncate a lock file a peer may hold.
        fp = open(self.lock_path, "a")
        fcntl.flock(fp.fileno(), fcntl.LOCK_EX)
        return fp

    @staticmethod
    def _release(fp: _LockFile) -> None:
        try:
            fcntl.flock(fp.fileno(), fcntl.LOCK_UN)
        finally:
            fp.close()

    def read_lines(self) -> list[dict[str, Any]]:
        """Read+parse all JSONL records. Caller must hold the lock."""
        if not self.path.exists():
            return []
        records: list[dict[str, Any]] = []
        with open(self.path, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line:
                    records.append(json.loads(line))
        return records

    def append_line(self, record: dict[str, Any]) -> None:
        """Append one JSON line. Caller must hold the lock."""
        with open(self.path, "a", encoding="utf-8") as f:
            f.write(json.dumps(record) + "\n")
            f.flush()
            os.fsync(f.fileno())

    def read_json(self) -> Any:
        """Read the whole file as one JSON document. Caller must hold the lock."""
        if not self.path.exists():
            return None
        with open(self.path, encoding="utf-8") as f:
            data = f.read().strip()
        if not data:
            return None
        return json.loads(data)

    def write_json(self, value: Any) -> None:
        """Atomically rewrite the whole file as one JSON document."""
        tmp = self.path.with_name(f"{self.path.name}.tmp.{os.getpid()}")
        try:
            with open(tmp, "w", encoding="utf-8") as f:
                json.dump(value, f, indent=2, sort_keys=True)
                f.write("\n")
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp, self.path)
        finally:
            if tmp.exists():
                tmp.unlink()


class EventLedger:
    """Append-only JSONL ledger of governed-write events (B1c).

    Records are immutable once written — no update/delete. ``record()``
    generates the id and timestamp and appends atomically under lock;
    ``records_for``/``all`` read the whole file under lock and return
    parsed :class:`GovernedWriteRecord` instances.
    """

    def __init__(self, path: Path):
        self._store = _LockedFile(Path(path))

    def record(
        self,
        writer: WriterIdentity,
        materials: dict[str, str],
        products: dict[str, str],
        worktree: str,
        commit: str,
        byproducts: dict[str, str] | None = None,
    ) -> GovernedWriteRecord:
        """Append a new governed-write record and return it."""
        entry = GovernedWriteRecord.new(
            writer=writer,
            materials=materials,
            products=products,
            worktree=worktree,
            commit=commit,
            byproducts=byproducts,
        )
        lock = self._store._acquire()
        try:
            self._store.append_line(_record_to_dict(entry))
        finally:
            self._store._release(lock)
        return entry

    def records_for(self, worktree: str, commit: str) -> list[GovernedWriteRecord]:
        """Return every record for ``(worktree, commit)``, insertion order."""
        lock = self._store._acquire()
        try:
            raw = self._store.read_lines()
        finally:
            self._store._release(lock)
        return [
            _dict_to_record(r)
            for r in raw
            if r["worktree"] == worktree and r["commit"] == commit
        ]

    def all(self) -> list[GovernedWriteRecord]:
        """Return every record in the ledger, insertion order."""
        lock = self._store._acquire()
        try:
            raw = self._store.read_lines()
        finally:
            self._store._release(lock)
        return [_dict_to_record(r) for r in raw]


class WriterRegistry:
    """JSON-file registry of governed-writer identities (B1c).

    ``registered()`` matches on the **(binary_hash, config_hash) pair only**
    (ADR-0017: identity is the hash pair, never the name) — two entries with
    the same name but different ``config_hash`` are distinct identities, and
    an identity whose config changed since registration is simply not found.

    # loosening: human-only per ADR-0009 — adding a writer newly ADMITS
    # deltas it did not admit before, which is a loosening of the
    # enforcement boundary. Phase-1 enforcement of WHO calls ``register`` is
    # the CLI's mandatory ``--by`` flag (``scripts/factory_enforce.py
    # register-writer``), not a code-level identity check here.
    """

    def __init__(self, path: Path):
        self._store = _LockedFile(Path(path))

    def register(self, identity: WriterIdentity, by: str) -> None:
        """Register ``identity``. Raises ``ValueError`` if ``by`` is empty."""
        if not by:
            raise ValueError(
                "register() requires a non-empty `by` (the human ratifying "
                "this loosening, per ADR-0009)"
            )
        lock = self._store._acquire()
        try:
            entries = self._store.read_json() or []
            entries.append(
                {
                    **_writer_to_dict(identity),
                    "registered_by": by,
                    "registered_at": _utc_now_iso(),
                }
            )
            self._store.write_json(entries)
        finally:
            self._store._release(lock)

    def registered(self, identity: WriterIdentity) -> bool:
        """Return whether ``identity``'s (binary_hash, config_hash) pair is
        registered."""
        lock = self._store._acquire()
        try:
            entries = self._store.read_json() or []
        finally:
            self._store._release(lock)
        return any(
            e["binary_hash"] == identity.binary_hash
            and e["config_hash"] == identity.config_hash
            for e in entries
        )

    def entries(self) -> list[WriterIdentity]:
        """Return every registered identity (introspection; not a CLI verb)."""
        lock = self._store._acquire()
        try:
            entries = self._store.read_json() or []
        finally:
            self._store._release(lock)
        return [_writer_from_dict(e) for e in entries]