orchestrata_ultimate.factory.anchors.store

Reference › orchestrata_ultimate › factory › anchors › store

store holds AnchorRegistry, the persistent store of semantic anchors (SPEC-033 row B15a, ADR-0013). The registry is one JSON file containing an array of anchor records (the docstring names factory_kb/anchors.json as the default CLI location). It follows the locking pattern of the assumption ledger: _LockedJsonFile takes an exclusive fcntl.flock on a sidecar lock file around every read-modify-write and rewrites the whole array through a temporary file and os.replace, so a crash cannot leave a torn file. The whole array is rewritten because each mutation needs the full set to check id collisions.

Anchors are constitution-tier and human-ratified, so the registry has no update method. propose stores an anchor with its status forced to PROPOSED; ratify moves a PROPOSED anchor to RATIFIED and records who ratified it; supersede moves a RATIFIED anchor to SUPERSEDED and links its replacement. Editing a ratified anchor means superseding it and proposing a new one. Every supersede appends an entry to a separate, never-rewritten weakening ledger (<path>.weakening-ledger.jsonl, written by _append_weakening_ledger) that records the old id, the new id and the person responsible. anchors lists records filtered by requirement or status and get finds one by id. Importers are gate, intake, source, the decision phase, the fast-path classifier and several factory CLIs.

At a glance

Class diagram

Dependency diagram

API

Classes

_LockedJsonFile

Line 75

Shared locked/atomic single-JSON-array file primitive.

AnchorRegistry

Line 129

JSON-file-backed store of :class:Anchor records.

Functions

Source

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

"""``AnchorRegistry`` -- the SPEC-033 semantic anchor store (B15a, ADR-0013).

A single JSON file (a JSON array of anchor records; default CLI location
``factory_kb/anchors.json``), atomic writes + ``fcntl`` locking mirroring
:mod:`orchestrata_ultimate.factory.assumptions.ledger`'s ``_LockedJsonlFile``:
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. Unlike the ledger's one-record-per-line JSONL, anchors are always
read/written as one full JSON-array snapshot, since the registry is small and
every mutation already needs the whole set to check id collisions.

Per ADR-0013, anchors are constitution-tier and human-ratified only: there is
**no update method**. The only way to change a RATIFIED anchor's meaning is
:meth:`AnchorRegistry.supersede` followed by a fresh :meth:`AnchorRegistry.
propose` -- any edit to a ratified anchor is a loosening by definition, and
every supersede appends a durable, never-rewritten entry to the
"anchor-weakened-to-pass" ledger (``<path>.weakening-ledger.jsonl``).
"""

from __future__ import annotations

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

from orchestrata_ultimate.factory.anchors.models import (
    Anchor,
    AnchorClass,
    AnchorStatus,
)

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


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


def _anchor_to_record(anchor: Anchor) -> dict[str, Any]:
    return {
        "anchor_id": anchor.anchor_id,
        "requirement_id": anchor.requirement_id,
        "klass": anchor.klass.value,
        "statement": anchor.statement,
        "boundary": anchor.boundary,
        "source": anchor.source,
        "status": anchor.status.value,
        "ratified_by": anchor.ratified_by,
        "ratified_at": anchor.ratified_at,
        "superseded_by": anchor.superseded_by,
    }


def _record_to_anchor(record: dict[str, Any]) -> Anchor:
    return Anchor(
        anchor_id=record["anchor_id"],
        requirement_id=record["requirement_id"],
        klass=AnchorClass(record["klass"]),
        statement=record["statement"],
        boundary=record.get("boundary", ""),
        source=record.get("source", ""),
        status=AnchorStatus(record["status"]),
        ratified_by=record.get("ratified_by", ""),
        ratified_at=record.get("ratified_at", ""),
        superseded_by=record.get("superseded_by"),
    )


class _LockedJsonFile:
    """Shared locked/atomic single-JSON-array file primitive.

    Not part of the public API -- ``AnchorRegistry`` holds one and builds its
    record (de)serialization + domain rules on top, mirroring the
    ``_LockedJsonlFile`` split in
    :mod:`orchestrata_ultimate.factory.assumptions.ledger`.
    """

    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:
        """Take the exclusive lock, blocking until it is available."""
        # '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:
        """Release a lock handle previously returned by :meth:`acquire`."""
        try:
            fcntl.flock(fp.fileno(), fcntl.LOCK_UN)
        finally:
            fp.close()

    def read_records(self) -> list[dict[str, Any]]:
        """Read+parse the JSON array. Caller must hold the lock."""
        if not self.path.exists():
            return []
        with open(self.path, encoding="utf-8") as f:
            content = f.read().strip()
        if not content:
            return []
        records: list[dict[str, Any]] = json.loads(content)
        return records

    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:
                json.dump(records, f, indent=2)
                f.flush()
                os.fsync(f.fileno())
            os.replace(tmp, self.path)
        finally:
            if tmp.exists():
                tmp.unlink()


class AnchorRegistry:
    """JSON-file-backed store of :class:`Anchor` records.

    ``propose``/``ratify``/``supersede``/``anchors``/``get`` is the entire
    public API -- there is deliberately no ``update``: per ADR-0013, editing a
    ratified anchor is only ever expressible as supersede + propose-new.
    """

    def __init__(self, path: Path):
        self._store = _LockedJsonFile(Path(path))
        self.path = self._store.path
        self.weakening_ledger_path = self.path.with_name(
            self.path.name + ".weakening-ledger.jsonl"
        )

    def _append_weakening_ledger(
        self, old_anchor_id: str, new_anchor_id: str, by: str
    ) -> None:
        """Append one never-rewritten entry recording an anchor weakening."""
        entry = {
            "old_anchor_id": old_anchor_id,
            "new_anchor_id": new_anchor_id,
            "by": by,
            "at": _utc_now_iso(),
        }
        with open(self.weakening_ledger_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(entry) + "\n")
            f.flush()
            os.fsync(f.fileno())

    def propose(self, anchor: Anchor) -> Anchor:
        """Persist ``anchor`` with status forced to ``PROPOSED``.

        Any ratification/supersession metadata on the input is cleared, so a
        fresh proposal always starts clean regardless of what was passed.

        Raises:
            ValueError: If an anchor with the same ``anchor_id`` already
                exists (deterministic-id collision -- the idempotency signal
                :func:`~orchestrata_ultimate.factory.anchors.intake.
                import_dor_candidates` relies on to skip already-proposed
                candidates).
        """
        forced = dataclasses.replace(
            anchor,
            status=AnchorStatus.PROPOSED,
            ratified_by="",
            ratified_at="",
            superseded_by=None,
        )
        lock = self._store.acquire()
        try:
            records = self._store.read_records()
            for record in records:
                if record["anchor_id"] == forced.anchor_id:
                    raise ValueError(
                        f"duplicate anchor id {forced.anchor_id!r} (same "
                        "requirement + class + normalized statement already "
                        "in the registry)"
                    )
            records.append(_anchor_to_record(forced))
            self._store.write_records(records)
            return forced
        finally:
            self._store.release(lock)

    def ratify(self, anchor_id: str, by: str) -> Anchor:
        """Transition a ``PROPOSED`` anchor to ``RATIFIED``.

        Ratification is the HUMAN act (ADR-0013) -- ``by`` is the ratifying
        human's identity and must be non-empty. This method does not itself
        distinguish a human caller from a model one; that guarantee is a
        process/CLI-boundary property, not something this file can enforce.

        Raises:
            ValueError: If ``by`` is empty, ``anchor_id`` is unknown, or the
                anchor is not currently ``PROPOSED``.
        """
        if not by:
            raise ValueError("by must not be empty")
        lock = self._store.acquire()
        try:
            records = self._store.read_records()
            index = next(
                (i for i, r in enumerate(records) if r["anchor_id"] == anchor_id),
                None,
            )
            if index is None:
                raise ValueError(f"unknown anchor id {anchor_id!r}")
            anchor = _record_to_anchor(records[index])
            if anchor.status != AnchorStatus.PROPOSED:
                raise ValueError(
                    f"cannot ratify anchor {anchor_id!r} in status "
                    f"{anchor.status.value!r} (only PROPOSED anchors can be "
                    "ratified)"
                )
            ratified = dataclasses.replace(
                anchor,
                status=AnchorStatus.RATIFIED,
                ratified_by=by,
                ratified_at=_utc_now_iso(),
            )
            records[index] = _anchor_to_record(ratified)
            self._store.write_records(records)
            return ratified
        finally:
            self._store.release(lock)

    def supersede(self, anchor_id: str, by_anchor_id: str, by: str) -> Anchor:
        """Transition a ``RATIFIED`` anchor to ``SUPERSEDED``.

        Per ADR-0013 this is the ONLY path by which a ratified anchor's
        meaning can change -- there is no update method. Appends one entry
        to the anchor-weakened-to-pass ledger.

        Raises:
            ValueError: If ``by`` or ``by_anchor_id`` is empty, ``anchor_id``
                is unknown, or the anchor is not currently ``RATIFIED``.
        """
        if not by:
            raise ValueError("by must not be empty")
        if not by_anchor_id:
            raise ValueError("by_anchor_id must not be empty")
        lock = self._store.acquire()
        try:
            records = self._store.read_records()
            index = next(
                (i for i, r in enumerate(records) if r["anchor_id"] == anchor_id),
                None,
            )
            if index is None:
                raise ValueError(f"unknown anchor id {anchor_id!r}")
            anchor = _record_to_anchor(records[index])
            if anchor.status != AnchorStatus.RATIFIED:
                raise ValueError(
                    f"cannot supersede anchor {anchor_id!r} in status "
                    f"{anchor.status.value!r} (only RATIFIED anchors can be "
                    "superseded)"
                )
            superseded = dataclasses.replace(
                anchor, status=AnchorStatus.SUPERSEDED, superseded_by=by_anchor_id
            )
            records[index] = _anchor_to_record(superseded)
            self._store.write_records(records)
            self._append_weakening_ledger(anchor_id, by_anchor_id, by)
            return superseded
        finally:
            self._store.release(lock)

    def anchors(
        self,
        requirement_id: str | None = None,
        status: AnchorStatus | None = None,
    ) -> list[Anchor]:
        """Return every anchor, optionally filtered by requirement/status."""
        lock = self._store.acquire()
        try:
            records = self._store.read_records()
        finally:
            self._store.release(lock)
        result = [_record_to_anchor(r) for r in records]
        if requirement_id is not None:
            result = [a for a in result if a.requirement_id == requirement_id]
        if status is not None:
            result = [a for a in result if a.status == status]
        return result

    def get(self, anchor_id: str) -> Anchor | None:
        """Return the anchor with ``anchor_id``, or ``None``."""
        lock = self._store.acquire()
        try:
            records = self._store.read_records()
        finally:
            self._store.release(lock)
        for record in records:
            if record["anchor_id"] == anchor_id:
                return _record_to_anchor(record)
        return None