symfonic.memory.backends.postgres_journal_source_write

postgres_journal_source_write

Validate an Agent FLUSH's exact source WRITE inside journal preparation.

bound_source

bound_source(prepared, flush)

The one prepared WRITE that flush is the canonical selected FLUSH of.

The FLUSH intent digest commits to accepted_write, so only the exact sealed source WRITE reproduces it. None is a FLUSH prepared without a source (a direct journal caller, or one prepared before CORE-199).

Source code in src/symfonic/memory/backends/postgres_journal_source_write.py

def bound_source(prepared, flush):
    """The one prepared WRITE that ``flush`` is the canonical selected FLUSH of.

    The FLUSH intent digest commits to ``accepted_write``, so only the exact
    sealed source WRITE reproduces it. ``None`` is a FLUSH prepared without a
    source (a direct journal caller, or one prepared before CORE-199).
    """
    matches = [
        write for write in prepared
        if write.verb is MemoryOperationVerb.WRITE and write.scope == flush.scope
        and operation_for_selected_write(
            flush.operation_id, write, flush.selected_manifest) == flush
    ]
    if len(matches) > 1:
        seal.refuse()
    return matches[0] if matches else None

require_source_write async

require_source_write(conn, cipher, expected, scope, phase_key, ordinal, candidate, body)

Fail closed on a bound source; legacy direct journal intents stay separate.

This validates the Agent's source at prepare time. The public graph-only admission gate remains closed and needs its own commit-time validation.

Source code in src/symfonic/memory/backends/postgres_journal_source_write.py

async def require_source_write(conn, cipher, expected, scope, phase_key, ordinal,
                               candidate, body):
    """Fail closed on a bound source; legacy direct journal intents stay separate.

    This validates the Agent's source at prepare time. The public graph-only
    admission gate remains closed and needs its own commit-time validation.
    """
    source = body.get("accepted_write")
    if (type(source) is not dict or set(source) != {"operation_id", "intent_digest"}
            or candidate.verb is not MemoryOperationVerb.FLUSH
            or not phase_key.endswith(":publication")):
        seal.refuse()
    write_phase = phase_key[:-len("publication")] + "write"
    key = seal.phase(write_phase, ordinal)
    stored = await codec.phase_row(conn, codec.OPERATIONS, expected, key)
    if stored is None:
        seal.refuse()
    write = seal.restore(
        cipher, [expected, write_phase, ordinal], stored["sealed"], scope,
    )
    if (write.verb is not MemoryOperationVerb.WRITE
            or write.operation_id != source["operation_id"]
            or write.intent_digest != source["intent_digest"]
            or not write.mutation_bound
            or candidate.selected_manifest is None):
        seal.refuse()
    row = await conn.fetchrow(
        "SELECT * FROM memory_operation_receipts "
        "WHERE tenant_id=$1 AND operation_id=$2 FOR SHARE",
        scope.tenant, write.operation_id,
    )
    if (row is None or row["state"] != "COMMITTED" or not _exact(row, write)
            or row["write_manifest"] is None):
        seal.refuse()
    document = row["write_manifest"]
    manifest = SelectedRecordManifest.from_document(
        json.loads(document) if isinstance(document, str) else document,
    )
    if (manifest != candidate.selected_manifest
            or any(claim.owner != scope for claim in manifest.claims)
            or operation_for_selected_write(candidate.operation_id, write, manifest)
            != candidate):
        seal.refuse()

source_reference

source_reference(source)

The sealed READY form of a bound source, or None when unbound.

Source code in src/symfonic/memory/backends/postgres_journal_source_write.py

def source_reference(source):
    """The sealed READY form of a bound source, or ``None`` when unbound."""
    return None if source is None else [source.operation_id, source.intent_digest]