symfonic.capabilities.memory.operation_journal

operation_journal

Durable caller operation binding for receipt-backed Agent memory.

CheckpointIncomplete dataclass

CheckpointIncomplete()

A partial checkpoint exists and requires operator reconciliation.

CheckpointMissing dataclass

CheckpointMissing()

No complete output is durably known for this phase.

ClaimAcquired dataclass

ClaimAcquired(owner: InvocationOwner, revision: int)

Only the atomic first insert may return this owner and revision zero.

ClaimExisting dataclass

ClaimExisting(revision: int)

Identical prior claim, deliberately without owner authority.

CompleteCheckpoint dataclass

CompleteCheckpoint(phase_key: str, ordinal: int, output_bytes: bytes, record_ids: tuple[str, ...], revision: int, selected_manifest: SelectedRecordManifest | None = None)

Immutable output and optional exact pending-row manifest for one phase.

InvocationOwner dataclass

InvocationOwner(invocation_id: str, full_scope: MemoryScope, request_digest: str, token: bytes)

First claimant's opaque authority; the adapter persists only its verifier.

Construction alone grants nothing: every mutation must compare the token's verifier, complete scope and request digest with the durable claim.

JournalUnknown dataclass

JournalUnknown()

Ambiguous persistence/read; no provider, tool, or mutation retry.

MemoryOperationJournalPort

Bases: Protocol

Persist exact phase identity and intent before the corresponding I/O.

Claim (tenant, invocation ID) against the full scope and request digest before any model call. A changed claim refuses. Prepare atomically binds (tenant, invocation ID, phase, ordinal) to one exact operation. Repeating equivalent intent returns that binding; a mismatch refuses. Persist an opaque collision-resistant operation ID bound to the tenant and a frozen canonical intent digest, never joined user strings or a mutable mapping. FLUSH binds the ordered authenticated pending manifest and exact IDs. Its full-owner claims and publication generations must equal the complete checkpoint manifest; an IDs-only checkpoint cannot authorize publication. For FLUSH, prepare must reject missing, duplicate, or differently ordered selected IDs unless they equal the complete checkpoint's record-ID tuple. Storage must survive restart and be shared across worker processes. Only an atomic first insert returns an unforgeable owner token and revision zero. Persist only its verifier; never reissue it or expire/reassign it. An identical repeat is Existing without owner; an ambiguous claim is UNKNOWN without owner. Every mutation verifies owner and expected revision with one atomic compare-and-swap. A complete output/record-ID checkpoint must precede prepare for the exact phase. The adapter must protect output bytes at rest and exclude output, token and digests from logs and errors. Missing, incomplete or UNKNOWN state on an existing claim never authorizes provider, tool or mutation replay; operator recovery or refusal is required. Manual cycles crossing invocations need a separate claim namespace. This protocol alone does not enable receipt-backed Agent composition.

An in-memory test double cannot satisfy the durability requirement.

PreparedOperation dataclass

PreparedOperation(operation: MemoryOperation, revision: int)

Persisted operation binding and the next owner CAS revision.

receipt_backed

receipt_backed(store: Any) -> bool

Detect known receipt ownership; unresolved wrapper paths refuse.

Opaque custom wrappers must expose a checked reference or receipt marker; this bounded walk cannot prove ownership hidden under arbitrary field names.

Source code in src/symfonic/capabilities/memory/operation_journal.py

def receipt_backed(store: Any) -> bool:
    """Detect known receipt ownership; unresolved wrapper paths refuse.

    Opaque custom wrappers must expose a checked reference or receipt marker;
    this bounded walk cannot prove ownership hidden under arbitrary field names.
    """
    pending: list[tuple[Any, frozenset[int]]] = [(store, frozenset())]
    seen: set[int] = set()
    while pending:
        current, parents = pending.pop()
        if current is None:
            continue
        marker = id(current)
        if marker in parents:
            return True  # An unresolved wrapper cycle must not admit mutations.
        if marker in seen:
            continue
        seen.add(marker)
        if len(seen) > 32:
            return True  # Bound unexpectedly deep or branching compositions.
        if _unresolved_forwarder(current):
            return True
        domain = _stored_reference(current, "transaction_domain")
        if domain is _UNRESOLVED or (domain is not _MISSING and domain is not None):
            return True
        lineage = parents | {marker}
        for name in _WRAPPER_REFERENCES:
            child = _stored_reference(current, name)
            if child is _MISSING:
                continue
            if child is _UNRESOLVED:
                key = (type(current).__module__, type(current).__name__, name)
                backing = _KNOWN_PROJECTIONS.get(key)
                if backing is None:
                    return True
                stored = _stored_reference(current, backing)
                if stored is _MISSING or stored is _UNRESOLVED:
                    return True
                continue  # This known property projects an inspected stored field.
            if child is current:
                return True
            if child is not None:
                pending.append((child, lineage))
    return False