Opt-in staged publication for the PostgreSQL v1 invocation journal.
READY is one journal unit: exact owner/revision, no accepted completion, the
durable claim fence installed, every WRITE receipt COMMITTED, and exactly one
prepared FLUSH whose receipt was never reserved. It seals the run identity and
the output digest, then advances the revision. Nothing is published. Since
CORE-339a it also seals a versioned per-record content digest of every staged
row (postgres_staged_content); a READY sealed without one never activates.
Source (CORE-199). When the staged FLUSH is an Agent FLUSH bound to its
accepted WRITE, READY also seals that WRITE's exact identity, and every
snapshot re-derives it from the sealed journal WRITE. Activation carries it
into the publication transaction, which re-reads the WRITE receipt there. A
READY that names a source the journal no longer holds refuses; it is never
read as unbound. A READY sealed without the key (before CORE-199) is unbound
unless its FLUSH still derives a source, which is then bound.
ACTIVE is not stored as a state: it is the staged FLUSH receipt being exactly
COMMITTED together with a matching postgres_staged_proof row, both of
which the receipt domain commits atomically with the graph release and the
vector writes. Absent receipt is READY; PENDING, inexact, or COMMITTED
without its exact proof is UNKNOWN. A proof is never backfilled from a receipt.
snapshot
async
snapshot(journal, invocation_id, scope, *, attest=None)
One read-only snapshot; scope or claim mismatch refuses, never MISSING.
Source code in src/symfonic/memory/backends/postgres_journal_staged.py
| async def snapshot(journal, invocation_id, scope, *, attest=None):
"""One read-only snapshot; scope or claim mismatch refuses, never MISSING."""
from symfonic.memory.backends.postgres_operation_receipt import _exact
probe = codec.identity(invocation_id, scope, "0" * 64)
async def work(conn):
if attest is not None:
await attest(conn)
found = await conn.fetchrow(
f"SELECT request_digest FROM {codec.JOURNAL} "
"WHERE tenant_id=$1 AND invocation_id=$2",
probe["tenant"], probe["invocation"],
)
if found is None:
return StagedSnapshot(StagedStatus.MISSING)
expected = codec.identity(invocation_id, scope, found["request_digest"])
claim = await codec.fetch(conn, expected)
if await completion_row(conn, expected) is not None:
seal.refuse() # an accepted CompletedTurn is not a staged turn
stored = await _row(conn, expected)
if stored is None:
return StagedSnapshot(StagedStatus.STAGED)
result, sealed_content, sealed_source = _restore(
journal, expected, claim, stored, scope)
prepared = await prepared_operations(conn, journal._cipher, expected, scope)
matches = [item for item in prepared
if (item.operation_id, item.intent_digest) == result.publication]
if len(matches) != 1:
seal.refuse()
operation = matches[0]
_flush(operation, scope)
source = _source(prepared, operation, sealed_source)
receipt = await _receipt(conn, operation)
if receipt is None:
status = StagedStatus.READY
elif (_exact(receipt, operation) and receipt["state"] == "COMMITTED"
and await proof.matches(conn, expected, stored, result, operation)):
status = StagedStatus.ACTIVE # a receipt alone, even exact, is UNKNOWN
else:
status = StagedStatus.UNKNOWN
return StagedSnapshot(status, result, operation, sealed_content,
hashlib.sha256(bytes(stored["sealed"])).hexdigest(), source)
try:
return await journal._unit(work, snapshot=True)
except MemoryContractError:
raise
except Exception:
journal._uncertain = True
return JournalUnknown()
|