symfonic.capabilities.memory.operation_receipt

operation_receipt

Public identity and reconciliation contract for durable memory operations.

An operation ID belongs to the caller and must survive a process restart. The caller persists the operation ID, verb, full scope and content digest before mutation. The intent digest binds that ID to a verb and complete kind-aware scope. COMMITTED confirms only the exact receipt; UNKNOWN never authorizes replay. This phase does not resolve UNKNOWN to ABORTED or provide a retry path. Selected Mongo publication is separate from the still-closed Agent admission.

MemoryOperation dataclass

MemoryOperation(operation_id: str, verb: MemoryOperationVerb, scope: MemoryScope, intent: Mapping[str, Any])

Caller-stable identity retaining only an immutable selected manifest.

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

def __init__(
    self,
    operation_id: str,
    verb: MemoryOperationVerb,
    scope: MemoryScope,
    intent: Mapping[str, Any],
) -> None:
    _validate_parts(operation_id, verb, scope)
    object.__setattr__(self, "operation_id", operation_id)
    object.__setattr__(self, "verb", verb)
    object.__setattr__(self, "scope", scope)
    frozen = _canonical(intent, set())
    manifest = None
    if "selected_manifest" in frozen:
        from symfonic.capabilities.memory.selected_records import SelectedRecordManifest

        manifest = SelectedRecordManifest.from_document(frozen["selected_manifest"])
        manifest.require_scope(scope)
        if tuple(frozen.get("selected_ids", ())) != manifest.record_ids:
            raise MemoryContractError("selected IDs do not match the manifest")
    object.__setattr__(self, "selected_manifest", manifest)
    object.__setattr__(self, "mutation_bound", True)
    content_digest = _content_digest(frozen)
    object.__setattr__(self, "content_digest", content_digest)
    object.__setattr__(self, "intent_digest", _bound_digest(verb, scope, content_digest))

from_intent classmethod

from_intent(operation_id: str, verb: MemoryOperationVerb, scope: MemoryScope, intent: Mapping[str, Any]) -> MemoryOperation

Build the same identity from equivalent canonical intent mappings.

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

@classmethod
def from_intent(
    cls,
    operation_id: str,
    verb: MemoryOperationVerb,
    scope: MemoryScope,
    intent: Mapping[str, Any],
) -> MemoryOperation:
    """Build the same identity from equivalent canonical intent mappings."""
    return cls(operation_id, verb, scope, intent)

restore classmethod

restore(operation_id: str, verb: MemoryOperationVerb, scope: MemoryScope, content_digest: str) -> MemoryOperation

Rebuild a persisted identity for reconciliation, without body replay.

The caller must persist these fields before attempting the mutation. Rebinding the content digest to the supplied verb and full scope keeps a restored identity from claiming another child scope's receipt.

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

@classmethod
def restore(
    cls,
    operation_id: str,
    verb: MemoryOperationVerb,
    scope: MemoryScope,
    content_digest: str,
) -> MemoryOperation:
    """Rebuild a persisted identity for reconciliation, without body replay.

    The caller must persist these fields before attempting the mutation.
    Rebinding the content digest to the supplied verb and full scope keeps a
    restored identity from claiming another child scope's receipt.
    """
    _validate_parts(operation_id, verb, scope)
    if (
        type(content_digest) is not str
        or len(content_digest) != 64
        or any(char not in "0123456789abcdef" for char in content_digest)
    ):
        raise ValueError("content digest must be lowercase SHA-256 hex")
    value = object.__new__(cls)
    object.__setattr__(value, "selected_manifest", None)
    object.__setattr__(value, "mutation_bound", False)
    object.__setattr__(value, "operation_id", operation_id)
    object.__setattr__(value, "verb", verb)
    object.__setattr__(value, "scope", scope)
    object.__setattr__(value, "content_digest", content_digest)
    object.__setattr__(value, "intent_digest", _bound_digest(verb, scope, content_digest))
    return value

MemoryOperationOutcome

Bases: StrEnum

A positive exact receipt, or a result that cannot authorize replay.

MemoryOperationOutcomeUnknown

MemoryOperationOutcomeUnknown(operation: MemoryOperation)

Bases: MemoryContractError

The caller must reconcile this exact identity; no body retry is safe.

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

def __init__(self, operation: MemoryOperation) -> None:
    self.operation = operation
    super().__init__(
        "memory operation outcome unknown; reconcile this identity; "
        "this contract never authorizes replay"
    )

MemoryOperationReconcilerPort

Bases: Protocol

Read-only outcome check; it never reserves an ID or replays a body.

MemoryOperationVerb

Bases: StrEnum

Reserved mutation families; values alone admit no lifecycle route.

require_operation

require_operation(operation: MemoryOperation | None, verb: MemoryOperationVerb, scope: MemoryScope) -> MemoryOperation

Require the caller's exact persisted identity before mutation I/O.

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

def require_operation(
    operation: MemoryOperation | None,
    verb: MemoryOperationVerb,
    scope: MemoryScope,
) -> MemoryOperation:
    """Require the caller's exact persisted identity before mutation I/O."""
    if type(operation) is not MemoryOperation:
        raise MemoryContractError("a receipt-backed mutation requires a caller MemoryOperation")
    if not operation.mutation_bound:
        raise MemoryContractError("restored operation is reconciliation-only")
    if operation.verb is not verb or operation.scope != scope:
        raise MemoryContractError("memory operation verb and full scope must match the mutation")
    return operation

validate_transaction_operation

validate_transaction_operation(domain: Any, operation: MemoryOperation | None, verb: MemoryOperationVerb, scope: MemoryScope) -> bool

Validate receipt identity before any cycle or lifecycle body runs.

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

def validate_transaction_operation(
    domain: Any,
    operation: MemoryOperation | None,
    verb: MemoryOperationVerb,
    scope: MemoryScope,
) -> bool:
    """Validate receipt identity before any cycle or lifecycle body runs."""
    receipt = callable(getattr(domain, "operation_transaction", None))
    if receipt:
        if type(scope) is not MemoryScope:
            raise MemoryContractError("receipt-backed mutation needs its exact MemoryScope")
        require_operation(operation, verb, scope)
    elif operation is not None:
        raise MemoryContractError("memory operation needs a receipt transaction domain")
    return receipt