Skip to content

symfonic.memory.backends.postgres_receipt_write

postgres_receipt_write

Atomic accepted WRITE claims, reconciled by exact caller operation identity.

PostgresReceiptWriter 

PostgresReceiptWriter(domain: Any, store: Any)

The public receipt Agent writer binds its actual records to the intent.

Source code in src/symfonic/memory/backends/postgres_receipt_write.py 
 95
 96
 97
 98
 99
100
101
102
103
def __init__(self, domain: Any, store: Any) -> None:
    from symfonic.capabilities.memory.graph_receipt import domain_of
    from symfonic.capabilities.memory.graph_store import GraphBackedHms
    from symfonic.memory.backends.postgres_receipt_views import receipt_pair_domain

    if (type(store) is not GraphBackedHms or domain_of(store) is not domain
            or receipt_pair_domain(store._graph, store._vectors) is not domain):
        raise MemoryContractError("WRITE requires the exact domain-owned HMS")
    self._domain, self._store = domain, store

ReceiptWrites

reconcile_write async 

reconcile_write(operation: MemoryOperation) -> Any

Read the exact committed manifest; UNKNOWN never adopts node IDs.

Historical receipts without a manifest remain unrecoverable. This does not replay a mutation or infer an aborted reservation.

Source code in src/symfonic/memory/backends/postgres_receipt_write.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
async def reconcile_write(self, operation: MemoryOperation) -> Any:
    """Read the exact committed manifest; UNKNOWN never adopts node IDs.

    Historical receipts without a manifest remain unrecoverable. This does
    not replay a mutation or infer an aborted reservation.
    """
    from symfonic.memory.backends.postgres_operation_receipt import _exact

    if (type(operation) is not MemoryOperation
            or operation.verb is not MemoryOperationVerb.WRITE):
        raise MemoryContractError("WRITE reconciliation requires an exact WRITE identity")
    conn = await self.pool.acquire()
    try:
        row = await conn.fetchrow(
            "SELECT * FROM memory_operation_receipts "
            "WHERE tenant_id=$1 AND operation_id=$2",
            operation.scope.tenant, operation.operation_id,
        )
        if (row is None or not _exact(row, operation) or row["state"] != "COMMITTED"
                or row["write_manifest"] is None):
            return None
        document = row["write_manifest"]
        manifest = SelectedRecordManifest.from_document(
            json.loads(document) if isinstance(document, str) else document,
        )
        manifest.require_scope(operation.scope)
        if any(claim.owner != operation.scope for claim in manifest.claims):
            raise MemoryContractError("WRITE manifest owner mismatch")
        return manifest
    finally:
        await self.pool.release(conn)