symfonic.memory.backends.mongo_operation_journal

mongo_operation_journal

Opt-in majority-durable Mongo Agent journal; no public Agent admission. Pass an authenticated encryption collaborator with encrypt(bytes)->bytes and decrypt(bytes)->bytes (for example Fernet/MultiFernet). Keys are externally owned; retain old decryption keys across rotation. Ambiguous storage fails closed.

MongoOperationJournal

MongoOperationJournal(uri: str, database: str, *, cipher)

One owned client, first-owner claim and atomic owner/revision CAS.

A complete checkpoint is encrypted with its claim and phase binding. Prepare persists an opaque operation and requires that exact checkpoint. Receipt reconciliation is separate: UNKNOWN never permits body replay.

Source code in src/symfonic/memory/backends/mongo_operation_journal.py

def __init__(self, uri: str, database: str, *, cipher) -> None:
    if not callable(getattr(cipher, "encrypt", None)) or not callable(
        getattr(cipher, "decrypt", None)
    ):
        codec.refuse()
    try:
        from motor.motor_asyncio import AsyncIOMotorClient
        from pymongo.read_concern import ReadConcern
        from pymongo.read_preferences import ReadPreference
        from pymongo.write_concern import WriteConcern
    except ImportError as exc:
        raise ImportError("Mongo journal requires: pip install symfonic-core[mongodb]") from exc

    self._client = AsyncIOMotorClient(uri, retryWrites=False, serverSelectionTimeoutMS=5000)
    self._database, self._cipher = database, cipher
    self._closed = False
    self._uncertain = False
    self._rows = self._client[database]["__symfonic_invocation_journal"].with_options(
        read_concern=ReadConcern("majority"),
        write_concern=WriteConcern(w="majority", j=True, wtimeout=5000),
        read_preference=ReadPreference.PRIMARY,
    )

has_invocation async

has_invocation(invocation_id, full_scope, request_digest)

Read-only exact-identity check before a host issues a recovery handle.

An existing ID with a different request is a binding refusal, not absence.

Source code in src/symfonic/memory/backends/mongo_operation_journal.py

async def has_invocation(self, invocation_id, full_scope, request_digest):
    """Read-only exact-identity check before a host issues a recovery handle.

    An existing ID with a different request is a binding refusal, not absence.
    """
    expected = codec.identity(invocation_id, full_scope, request_digest)
    return await self._read(expected) is not None

verify_owner async

verify_owner(owner, expected_revision)

Majority-read exact ownership; no write retry or revision advance.

Source code in src/symfonic/memory/backends/mongo_operation_journal.py

async def verify_owner(self, owner, expected_revision):
    """Majority-read exact ownership; no write retry or revision advance."""
    from symfonic.capabilities.memory.owner_verifier import OwnerVerified

    expected = self._owner_binding(owner, expected_revision)
    try:
        await self._owned(expected, allow_completed=True)
        return OwnerVerified()
    except MemoryContractError:
        raise
    except Exception:
        self._uncertain = True
        return JournalUnknown()