symfonic.capabilities.memory.receipt_composition

receipt_composition

Receipt-only Agent memory stages, bound to one durable invocation scope.

ReceiptMemoryCapability dataclass

ReceiptMemoryCapability(read: MemoryCapability, writer: Any, scope: Any, journal: Any, records_from: Any, _seal: object, accepted_completion: Any = None, extractor: Any = None)

One guarded recall and one authenticated write/publication stage.

bind_quick async

bind_quick(coordinator: Any, *, policy_digest: str) -> None

Opt in this issued PostgreSQL receipt scope to graph-only durable QUICK.

Await before serving turns. Pin the complete phase configuration with the same SHA-256 digest across restarts. This does not admit raw PG consolidation or change its production erasure requirements.

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

async def bind_quick(self, coordinator: Any, *, policy_digest: str) -> None:
    """Opt in this issued PostgreSQL receipt scope to graph-only durable QUICK.

    Await before serving turns. Pin the complete phase configuration with
    the same SHA-256 digest across restarts. This does not admit raw PG
    consolidation or change its production erasure requirements.
    """
    if not issued_receipt(self):
        raise ConfigurationError("receipt memory requires validated factory composition")
    if not _postgres_journal(self.journal) or self.accepted_completion is None:
        raise ConfigurationError("durable receipt QUICK requires native PostgreSQL")
    await self.accepted_completion.bind_quick(coordinator, policy_digest=policy_digest)

recover_quick async

recover_quick(*, limit: int = 500) -> tuple[int, bool]

Advance one bounded accepted-turn batch without running turn bodies.

Repeat while the returned backlog flag is true. UNKNOWN stops recovery; reconcile its pending receipt before continuing or serving more turns.

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

async def recover_quick(self, *, limit: int = 500) -> tuple[int, bool]:
    """Advance one bounded accepted-turn batch without running turn bodies.

    Repeat while the returned backlog flag is true. UNKNOWN stops recovery;
    reconcile its pending receipt before continuing or serving more turns.
    """
    if not issued_receipt(self):
        raise ConfigurationError("receipt memory requires validated factory composition")
    if not _postgres_journal(self.journal) or self.accepted_completion is None:
        raise ConfigurationError("durable receipt QUICK requires native PostgreSQL")
    return await self.accepted_completion.recover_quick(limit=limit)

receipt_for_scope

receipt_for_scope(factory: Any, scope: MemoryScope) -> ReceiptMemoryCapability

Refuse every unsupported collaborator before exposing receipt stages.

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

def receipt_for_scope(factory: Any, scope: MemoryScope) -> ReceiptMemoryCapability:
    """Refuse every unsupported collaborator before exposing receipt stages."""
    store, journal = factory._store, factory._operation_journal
    native_writer = None
    accepted_completion = None
    if _postgres_journal(journal):
        _require_postgres(store, scope, journal)
        native_writer = journal._domain.writer_for(store)
        from symfonic.memory.backends.postgres_accepted_handoff import PostgresAcceptedHandoff

        accepted_completion = PostgresAcceptedHandoff(journal, journal._domain, scope)
    elif scope.identity_version == 1:
        native_writer = _require_native_mongo(store, scope, journal)
    else:
        _require_mongo(store, scope, journal)
    if (
        any(
            value is not None
            for value in (
                factory._summary,
                factory._reflection,
                factory._consolidation,
                factory._conversation,
                factory._activation,
                factory._retrieval_policy,
                factory._retrieval,
                factory._episodic_telemetry_sink,
            )
        )
        or factory._recent_turns != 0
        or factory._exclude_speakers != frozenset()
    ):
        raise ConfigurationError("receipt Agent has unsupported optional memory stage")
    if factory._extractor is not None:
        from symfonic.capabilities.memory.receipt_extraction import require_extractor

        require_extractor(factory._extractor, staged=_postgres_journal(journal)
                          and staged_journal(journal))
    activation = None
    if _postgres_journal(journal) and journal.linked_selection_acknowledgement:
        from symfonic.capabilities.memory.activation import SpreadingActivation
        from symfonic.capabilities.memory.graph_associations import GraphAssociationSource
        from symfonic.capabilities.memory.linked_spreading_policy import LinkedSpreadingPolicy

        admission = ReadAdmission((scope.path,))
        source = GraphAssociationSource(None, scope, admission, linked_edges=journal.linked_edges)
        activation = SpreadingActivation(source, linked_policy=LinkedSpreadingPolicy())
    read = MemoryCapability(
        hydrator_for(store, activation=activation),
        scope=scope,
        limit=factory._limit,
        recall_budget=factory._recall_budget,
        read_admission=ReadAdmission((scope.path,)),
    )
    capability = ReceiptMemoryCapability(
        read, native_writer or store, scope, journal,
        _delivered_records(producer_for(scope)), _FACTORY_SEAL,
        accepted_completion, factory._extractor,
    )
    _ISSUED.add(capability)
    return capability