Skip to content

symfonic.memory.backends.postgres_staged_activation

postgres_staged_activation

Verified, idempotent activation of one READY staged PostgreSQL v1 turn.

Activation runs the exact sealed FLUSH through the receipt-bound store, so the receipt domain commits the FLUSH receipt, the claim-fenced graph release and the vector writes in one transaction. It runs only after the committed-receipt reader bound at publisher composition returns a local final receipt whose owner scope, invocation, run and output digest equal the sealed READY. ACTIVE returns unchanged; MISSING, STAGED and UNKNOWN refuse and are never retried; a receipt that is absent publishes nothing; a reader that raises publishes nothing. Only after the receipt verifies is this task's FLUSH armed, so the receipt domain writes the postgres_staged_proof row in that same transaction; ACTIVE needs it, and the arming is cleared however flush ends.

Content (CORE-339a). The armed proof carries the READY's sealed per-record content digests. The receipt domain spends the FLUSH id inside the publication transaction and proves every sealed row present and unchanged under a row lock before any vector is written. A missing or rewritten row, or any other rollback, leaves no receipt in any state: the turn stays READY and activation raises, so it never reads UNKNOWN on its account. The raise is typed: a sealed row absent or rewritten is StagedContentRefused, which a retry repeats; any other rollback (an embedder timeout, a cancelled statement, a deadlock) is StagedActivationRolledBack, and the same activate call is safe to retry. A READY sealed before CORE-339a carries no content seal and refuses to activate with StagedContentRefused.

Source (CORE-199). A staged Agent FLUSH bound to its accepted WRITE is published under that same binding: the snapshot carries the exact WRITE the READY sealed, a preflight refuses before the FLUSH id is spent when its COMMITTED receipt is gone or no longer this manifest, and the publication transaction re-reads that receipt under a share lock before any row is touched. A source that changes after the preflight rolls the publication back whole and raises SourceWriteRefused, which a retry repeats. A FLUSH staged without a source (a direct journal caller, or a READY prepared before CORE-199) publishes unbound, as before.

Trust boundary. The publisher is a trusted-host contract, not a defence against hostile code in the publisher process: such code already holds the store and can flush it directly. The database half of a separate publisher is postgres_staged_roles; process and credential separation is outside both.

PostgresStagedPublisher 

PostgresStagedPublisher(journal, store, full_scope, receipts)

One staged journal, its atomic store, one exact scope, one trusted reader.

Composition is the only place a reader is accepted; activate takes the invocation id alone, so no call site can substitute a verifier.

Source code in src/symfonic/memory/backends/postgres_staged_activation.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
def __init__(self, journal, store, full_scope, receipts) -> None:
    from symfonic.memory.backends.postgres_operation_journal import (
        PostgresOperationJournal,
    )

    if type(journal) is not PostgresOperationJournal:
        seal.refuse()
    staged.require_staged(journal)
    require_atomic(journal, store)
    if type(full_scope) is not MemoryScope or full_scope.identity_version != 1:
        seal.refuse()
    full_scope.validate()
    read = getattr(receipts, "read_committed_final", None)
    if not callable(read) or receipts is journal or receipts is store:
        seal.refuse()
    self._journal, self._store, self._scope, self._read = (
        journal, store, full_scope, read)

activate async 

activate(invocation_id, *, _snapshot=None)

Publish one READY turn of the bound scope after a fresh exact read.

Source code in src/symfonic/memory/backends/postgres_staged_activation.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def activate(self, invocation_id, *, _snapshot=None):
    """Publish one READY turn of the bound scope after a fresh exact read."""
    journal, store, scope = self._journal, self._store, self._scope
    snapshot = _snapshot or staged.snapshot
    current = await snapshot(journal, invocation_id, scope)
    if type(current) is JournalUnknown or current.status is StagedStatus.ACTIVE:
        return current
    if current.status is not StagedStatus.READY:
        seal.refuse()  # MISSING, STAGED and UNKNOWN never authorize publication
    sealed = content.require_seal(current.content, current.sealed_sha256, current.operation)
    # The reader is given the key, never READY: it has nothing to echo.
    receipt = await self._read(current.ready.invocation_id, scope)
    if receipt is None:
        return current  # no committed local final turn, or it rolled back
    require_receipt(current.ready, receipt)
    domain = journal._domain
    source = current.source
    if source is not None:
        # Advisory, like the proof preflight: refuse before the id is spent.
        require_selected_write(current.operation, source, await _source_manifest(
            domain, source))
    # Drift refuses here, before the FLUSH id is spent and could only read UNKNOWN.
    await proof.preflight(domain.pool)
    # Armed only now: the proof is written in the FLUSH receipt transaction
    # for this exact operation, and nowhere else.
    domain.arm_staged_proof(proof.StagedProof(current.operation, current.ready, sealed))
    failure = None
    try:
        # A concurrent activation may own the id; the fresh snapshot decides.
        with (nullcontext() if source is None
              else bound_write_selected(domain, current.operation, source)):
            await store.flush(scope, operation=current.operation)
    except MemoryOperationOutcomeUnknown as exc:
        failure = exc
    finally:
        domain.disarm_staged_proof()
    after = await snapshot(journal, invocation_id, scope)
    if type(after) is not JournalUnknown and (
            after.ready != current.ready or after.operation != current.operation
            or after.source != current.source):
        seal.refuse()
    if failure is not None and type(after) is not JournalUnknown \
            and after.status is StagedStatus.READY:
        _refuse_rolled_back(failure)
    return after

StagedActivationRolledBack

Bases: MemoryContractError

Activation rolled back whole for a reason other than sealed content.

The turn is READY with no FLUSH receipt in any state, no proof and no vector, so calling activate again for the same invocation is safe.

require_atomic 

require_atomic(journal, store) -> None

Graph release and vector writes must share the journal's receipt domain.

Source code in src/symfonic/memory/backends/postgres_staged_activation.py 
73
74
75
76
77
78
79
80
81
82
83
84
def require_atomic(journal, store) -> None:
    """Graph release and vector writes must share the journal's receipt domain."""
    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

    domain = journal._domain
    if (type(store) is not GraphBackedHms or store._vectors is None
            or store._recall is None or domain_of(store) is not domain):
        seal.refuse()
    if receipt_pair_domain(store._graph, store._vectors) is not domain:
        seal.refuse()

require_receipt 

require_receipt(ready, receipt) -> None

Every field of the host's durable receipt equals the sealed READY.

Source code in src/symfonic/memory/backends/postgres_staged_activation.py 
87
88
89
90
91
92
93
94
def require_receipt(ready, receipt) -> None:
    """Every field of the host's durable receipt equals the sealed READY."""
    if (type(receipt) is not LocalFinalReceipt
            or receipt.full_scope != ready.full_scope
            or receipt.invocation_id != ready.invocation_id
            or receipt.run_id != ready.run_id
            or not secrets.compare_digest(receipt.output_sha256, ready.output_sha256)):
        seal.refuse()