orchestrata_ultimate.factory.assumptions.ledger

Reference › orchestrata_ultimate › factory › assumptions › ledger

ledger is the persistence layer for the assumption subsystem (SPEC-033 REQ-033-C1, Phase 0). It stores two things as JSONL files: the assumption ledger and the Project knowledge base of promoted adjudications. It is imported by cascade, dor, sources, the decision phase, the C1 bridge detector and three factory CLIs, and tested in tests/factory/test_ledger.py.

_LockedJsonlFile is the shared primitive. It takes an exclusive fcntl.flock on a sidecar lock file around every read-modify-write, and rewrites go through a sibling temporary file and os.replace, so a crash never leaves a torn file and concurrent writers serialise on the lock. AssumptionLedger builds on it: add rejects a duplicate id with ValueError, update rewrites a record in place, get and assumptions read back, and add_disclosure and disclosures store the interpretation disclosures of AC-033-C1-03. Each line carries a kind tag of assumption or disclosure.

KBProtocol is the knowledge-base seam (lookup, promote, all, supersede); its docstring notes it is file-backed today and graph-backed at B4a. FileProjectKB implements it over adjudications.jsonl. lookup returns a non-superseded match by exact fingerprint or by token-set Jaccard similarity at or above KB_FUZZY_MATCH_THRESHOLD (0.8). promote assigns the next sequential ADJ-NNNNNN id, and supersede marks an entry as replaced and raises if the id is unknown. The private _record_to_* and _*_to_record helpers convert the frozen dataclasses from models to and from JSON.

At a glance

Class diagram

Dependency diagram

API

Classes

_LockedJsonlFile

Line 57

Shared locked/atomic JSONL primitive for the ledger and the KB.

AssumptionLedger

Decorators: traces(spec='033', reqs=['REQ-033-C1']) · Line 211

File-based (JSONL) ledger of assumptions + interpretation disclosures.

KBProtocol (protocol)

Bases: Protocol · Line 309

The Project-KB seam — file-backed today, graph-backed at B4a.

FileProjectKB

Decorators: traces(spec='033', reqs=['REQ-033-C1']) · Line 330

JSONL-backed KBProtocol implementation (adjudications.jsonl).

Functions

Source

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

"""``AssumptionLedger`` + ``ProjectKB`` (SPEC-033 REQ-033-C1, Phase 0).

Both stores are JSONL files with atomic + locked I/O mirroring the pattern in
``orchestrata_ultimate.session.context.ContextFileManager``: 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 — exercised under free-threaded 3.14t in
``tests/factory/test_ledger.py``.

The ledger stores one JSON record per line, each tagged
``{"kind": "assumption" | "disclosure", ...}``. The KB stores one
``Adjudication`` record per line.
"""

from __future__ import annotations

import fcntl
import json
import os
from pathlib import Path
from typing import IO, Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.factory.assumptions.models import (
    Adjudication,
    Assumption,
    AssumptionStatus,
    InterpretationDisclosure,
    Pinning,
    PinningKind,
    Ruling,
    normalize,
)

# Fuzzy KB match: token-set Jaccard similarity of normalized assumption text
# must reach this threshold for FileProjectKB.lookup to treat two assumptions
# as "the same question" absent an exact fingerprint match.
KB_FUZZY_MATCH_THRESHOLD = 0.8

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


def _tokenize_normalized(text: str) -> set[str]:
    """Token set of a normalized string, for Jaccard similarity."""
    return set(normalize(text).split())


def _jaccard(a: set[str], b: set[str]) -> float:
    if not a or not b:
        return 0.0
    return len(a & b) / len(a | b)


class _LockedJsonlFile:
    """Shared locked/atomic JSONL primitive for the ledger and the KB.

    Not part of the public API — ``AssumptionLedger`` and ``FileProjectKB``
    each hold one and build their record (de)serialization on top.
    """

    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_records(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_record(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 write_records(self, records: list[dict[str, Any]]) -> None:
        """Atomically rewrite the whole file. Caller must hold the lock."""
        tmp = self.path.with_name(f"{self.path.name}.tmp.{os.getpid()}")
        try:
            with open(tmp, "w", encoding="utf-8") as f:
                for record in records:
                    f.write(json.dumps(record) + "\n")
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp, self.path)
        finally:
            if tmp.exists():
                tmp.unlink()


def _pinning_to_dict(pinning: Pinning | None) -> dict[str, Any] | None:
    if pinning is None:
        return None
    return {"kind": pinning.kind.value, "ref": pinning.ref, "excerpt": pinning.excerpt}


def _pinning_from_dict(data: dict[str, Any] | None) -> Pinning | None:
    if data is None:
        return None
    return Pinning(
        kind=PinningKind(data["kind"]), ref=data["ref"], excerpt=data["excerpt"]
    )


def _assumption_to_record(assumption: Assumption) -> dict[str, Any]:
    return {
        "kind": "assumption",
        "id": assumption.id,
        "text": assumption.text,
        "design_node": assumption.design_node,
        "author_identity": assumption.author_identity,
        "requirement_id": assumption.requirement_id,
        "status": assumption.status.value,
        "pinning": _pinning_to_dict(assumption.pinning),
        "stage_resolved": assumption.stage_resolved,
        "questions": list(assumption.questions),
        "created_at": assumption.created_at,
    }


def _record_to_assumption(record: dict[str, Any]) -> Assumption:
    return Assumption(
        id=record["id"],
        text=record["text"],
        design_node=record["design_node"],
        author_identity=record["author_identity"],
        requirement_id=record.get("requirement_id"),
        status=AssumptionStatus(record["status"]),
        pinning=_pinning_from_dict(record.get("pinning")),
        stage_resolved=record.get("stage_resolved"),
        questions=list(record.get("questions", [])),
        created_at=record["created_at"],
    )


def _disclosure_to_record(disclosure: InterpretationDisclosure) -> dict[str, Any]:
    return {
        "kind": "disclosure",
        "design_node": disclosure.design_node,
        "chosen_reading": disclosure.chosen_reading,
        "alternatives": list(disclosure.alternatives),
        "author_identity": disclosure.author_identity,
        "created_at": disclosure.created_at,
    }


def _record_to_disclosure(record: dict[str, Any]) -> InterpretationDisclosure:
    return InterpretationDisclosure(
        design_node=record["design_node"],
        chosen_reading=record["chosen_reading"],
        alternatives=list(record.get("alternatives", [])),
        author_identity=record["author_identity"],
        created_at=record["created_at"],
    )


def _adjudication_to_record(adjudication: Adjudication) -> dict[str, Any]:
    return {
        "id": adjudication.id,
        "fingerprint": adjudication.fingerprint,
        "assumption_text": adjudication.assumption_text,
        "ruling": adjudication.ruling.value,
        "pinning": _pinning_to_dict(adjudication.pinning),
        "adjudicator": adjudication.adjudicator,
        "requirement_id": adjudication.requirement_id,
        "adjudicated_at": adjudication.adjudicated_at,
        "superseded_by": adjudication.superseded_by,
    }


def _record_to_adjudication(record: dict[str, Any]) -> Adjudication:
    return Adjudication(
        id=record["id"],
        fingerprint=record["fingerprint"],
        assumption_text=record["assumption_text"],
        ruling=Ruling(record["ruling"]),
        pinning=_pinning_from_dict(record.get("pinning")),
        adjudicator=record["adjudicator"],
        requirement_id=record.get("requirement_id"),
        adjudicated_at=record["adjudicated_at"],
        superseded_by=record.get("superseded_by"),
    )


@traces(spec="033", reqs=["REQ-033-C1"])
class AssumptionLedger:
    """File-based (JSONL) ledger of assumptions + interpretation disclosures.

    One record per line, atomic + locked as described in the module
    docstring. Assumption identity is the deterministic ``Assumption.id``
    (fingerprint + design node); adding a duplicate raises ``ValueError``.
    """

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

    def add(self, assumption: Assumption) -> None:
        """Append a new assumption. Raises ``ValueError`` on duplicate id."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
            for record in records:
                if record["kind"] == "assumption" and record["id"] == assumption.id:
                    raise ValueError(
                        f"duplicate assumption id {assumption.id!r} "
                        "(same fingerprint + design_node already in ledger)"
                    )
            self._store.append_record(_assumption_to_record(assumption))
        finally:
            self._store._release(lock)

    def add_disclosure(self, disclosure: InterpretationDisclosure) -> None:
        """Append an interpretation disclosure (AC-033-C1-03)."""
        lock = self._store._acquire()
        try:
            self._store.append_record(_disclosure_to_record(disclosure))
        finally:
            self._store._release(lock)

    def get(self, assumption_id: str) -> Assumption | None:
        """Return the assumption with ``assumption_id``, or ``None``."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
        finally:
            self._store._release(lock)
        for record in records:
            if record["kind"] == "assumption" and record["id"] == assumption_id:
                return _record_to_assumption(record)
        return None

    def update(self, assumption: Assumption) -> None:
        """Rewrite the stored record for ``assumption.id`` in place.

        Raises:
            ValueError: If no assumption with that id is in the ledger yet.
        """
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
            new_record = _assumption_to_record(assumption)
            found = False
            rewritten: list[dict[str, Any]] = []
            for record in records:
                if record["kind"] == "assumption" and record["id"] == assumption.id:
                    rewritten.append(new_record)
                    found = True
                else:
                    rewritten.append(record)
            if not found:
                raise ValueError(
                    f"cannot update unknown assumption id {assumption.id!r}"
                )
            self._store.write_records(rewritten)
        finally:
            self._store._release(lock)

    def assumptions(self) -> list[Assumption]:
        """Return every assumption in the ledger, in insertion order."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
        finally:
            self._store._release(lock)
        return [_record_to_assumption(r) for r in records if r["kind"] == "assumption"]

    def disclosures(
        self, design_node: str | None = None
    ) -> list[InterpretationDisclosure]:
        """Return interpretation disclosures, optionally filtered by node."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
        finally:
            self._store._release(lock)
        result = [
            _record_to_disclosure(r) for r in records if r["kind"] == "disclosure"
        ]
        if design_node is not None:
            result = [d for d in result if d.design_node == design_node]
        return result


class KBProtocol(Protocol):
    """The Project-KB seam — file-backed today, graph-backed at B4a."""

    def lookup(self, fingerprint: str, text: str) -> Adjudication | None:
        """Return a non-superseded adjudication matching ``fingerprint``/``text``."""
        ...

    def promote(self, adjudication: Adjudication) -> Adjudication:
        """Assign an id to ``adjudication`` and persist it into the KB."""
        ...

    def all(self) -> list[Adjudication]:
        """Return every adjudication in the KB (including superseded ones)."""
        ...

    def supersede(self, adj_id: str, by_id: str) -> None:
        """Mark the adjudication ``adj_id`` as superseded by ``by_id``."""
        ...


@traces(spec="033", reqs=["REQ-033-C1"])
class FileProjectKB:
    """JSONL-backed ``KBProtocol`` implementation (``adjudications.jsonl``).

    ``lookup`` matches on exact fingerprint equality OR token-set Jaccard
    similarity of the normalized texts >= ``KB_FUZZY_MATCH_THRESHOLD``.
    Superseded adjudications (``superseded_by is not None``) never match.
    """

    def __init__(self, dir_path: Path):
        self.dir_path = Path(dir_path)
        self.dir_path.mkdir(parents=True, exist_ok=True)
        self._store = _LockedJsonlFile(self.dir_path / "adjudications.jsonl")

    def lookup(self, fingerprint: str, text: str) -> Adjudication | None:
        """Return the best non-superseded match for ``fingerprint``/``text``."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
        finally:
            self._store._release(lock)
        candidates = [
            _record_to_adjudication(r)
            for r in records
            if r.get("superseded_by") is None
        ]

        for adj in candidates:
            if adj.fingerprint == fingerprint:
                return adj

        query_tokens = _tokenize_normalized(text)
        best: Adjudication | None = None
        best_score = 0.0
        for adj in candidates:
            score = _jaccard(query_tokens, _tokenize_normalized(adj.assumption_text))
            if score >= KB_FUZZY_MATCH_THRESHOLD and score > best_score:
                best = adj
                best_score = score
        return best

    def promote(self, adjudication: Adjudication) -> Adjudication:
        """Assign the next sequential ``ADJ-NNNNNN`` id and append to the KB."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
            max_seq = 0
            for r in records:
                rid = r.get("id", "")
                if rid.startswith("ADJ-"):
                    try:
                        max_seq = max(max_seq, int(rid[4:]))
                    except ValueError:
                        continue
            new_id = f"ADJ-{max_seq + 1:06d}"
            promoted = Adjudication(
                id=new_id,
                fingerprint=adjudication.fingerprint,
                assumption_text=adjudication.assumption_text,
                ruling=adjudication.ruling,
                pinning=adjudication.pinning,
                adjudicator=adjudication.adjudicator,
                requirement_id=adjudication.requirement_id,
                adjudicated_at=adjudication.adjudicated_at,
                superseded_by=None,
            )
            self._store.append_record(_adjudication_to_record(promoted))
            return promoted
        finally:
            self._store._release(lock)

    def all(self) -> list[Adjudication]:
        """Return every adjudication in the KB, including superseded ones."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
        finally:
            self._store._release(lock)
        return [_record_to_adjudication(r) for r in records]

    def supersede(self, adj_id: str, by_id: str) -> None:
        """Mark ``adj_id`` as superseded by ``by_id``. Raises if unknown."""
        lock = self._store._acquire()
        try:
            records = self._store.read_records()
            found = False
            rewritten: list[dict[str, Any]] = []
            for record in records:
                if record["id"] == adj_id:
                    record = dict(record)
                    record["superseded_by"] = by_id
                    found = True
                rewritten.append(record)
            if not found:
                raise ValueError(f"cannot supersede unknown adjudication id {adj_id!r}")
            self._store.write_records(rewritten)
        finally:
            self._store._release(lock)