symfonic.memory.backends.postgres_edge_publication_writer

postgres_edge_publication_writer

Explicit native edge publication writer; defaults and fleet admission stay separate.

Requires preinstalled CORE-356 guards and the explicitly selected edge index. No backfill or implicit republication. Privileged function/DDL replacement still requires a fleet barrier; catalog checks cannot authenticate operator authority.

PostgresEdgePublicationsV1 dataclass

PostgresEdgePublicationsV1(pool: object, schema: str, edge_upsert_mode: str)

Standalone opt-in service; every returned identity follows its own commit.

Borrowed transactions are refused, including ambient receipt transactions. This port does not enlist an invocation receipt or authorize linked ack.

add_edge_v1 async

add_edge_v1(scope, edge) -> EdgePublicationV1

Publish a new edge, or return an existing same-identity publication.

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

async def add_edge_v1(self, scope, edge) -> EdgePublicationV1:
    """Publish a new edge, or return an existing same-identity publication."""
    return await self._insert(scope, edge, upsert=False)

update_owned_edge_v1 async

update_owned_edge_v1(scope, edge_id, props, *, expected) -> EdgePublicationV1

Full-owner/incarnation/revision CAS, including unchanged and uses-only updates.

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

async def update_owned_edge_v1(self, scope, edge_id, props, *, expected) -> EdgePublicationV1:
    """Full-owner/incarnation/revision CAS, including unchanged and uses-only updates."""
    _scope(scope)
    if (type(expected) is not EdgePublicationV1 or expected.owner != scope.identity
            or expected.tenant_id != scope.tenant_id or expected.edge_id != str(edge_id)
            or type(expected.incarnation) is not UUID or type(expected.revision) is not int
            or expected.revision < 1):
        raise MemoryContractError("edge publication requires exact owned receipt")
    if type(props) is not dict or not props.keys() <= _MUTABLE:
        raise MemoryContractError("edge publication update contains unsupported fields")
    async with self._transaction() as (conn, table):
        endpoints = MemoryEdge(id=expected.edge_id, tenant_id=scope.tenant_id,
                               source=expected.source, target=expected.target,
                               relationship=expected.relationship)
        await lock_native_endpoints(conn, scope, endpoints)
        row = await conn.fetchrow(
            f'SELECT * FROM {table} WHERE tenant_id=$1 AND id=$2 '
            'AND tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2 '
            'AND scope_path COLLATE "C"=$3 AND publication_format=1 '
            'AND publication_incarnation=$4 AND publication_revision=$5 FOR UPDATE',
            scope.tenant_id, str(edge_id), scope.scope_path,
            expected.incarnation, expected.revision)
        if row is None or _receipt(row) != expected:
            raise MemoryContractError("edge publication receipt is stale or mismatched")
        values = dict(row)
        episodic = values["source_episodic_ids"]
        values["source_episodic_ids"] = (
            json.loads(episodic) if isinstance(episodic, str) else episodic)
        values.update(props)
        updated = _edge(MemoryEdge.model_validate(values))
        row = await conn.fetchrow(f"""
            UPDATE {table} SET weight=$6, uses=$7, provenance=$8,
                source_episodic_ids=$9::jsonb, created_at=$10,
                publication_revision=publication_revision + CASE WHEN
                    ROW(weight,provenance,source_episodic_ids,created_at)
                    IS DISTINCT FROM ROW($6,$8,$9::jsonb,$10) THEN 1 ELSE 0 END
            WHERE tenant_id=$1 AND id=$2 AND tenant_id COLLATE "C"=$1 AND id COLLATE "C"=$2
                AND scope_path COLLATE "C"=$3 AND publication_format=1
                AND publication_incarnation=$4 AND publication_revision=$5 RETURNING *
            """, scope.tenant_id, str(edge_id), scope.scope_path,
            expected.incarnation, expected.revision, updated.weight, updated.uses,
            updated.provenance, json.dumps(updated.source_episodic_ids), updated.created_at)
        receipt = _receipt(row)
    return receipt

upsert_edge_v1 async

upsert_edge_v1(scope, edge) -> EdgePublicationV1

Publish or strengthen; refuse a float8 increment without finite progress.

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

async def upsert_edge_v1(self, scope, edge) -> EdgePublicationV1:
    """Publish or strengthen; refuse a float8 increment without finite progress."""
    return await self._insert(scope, edge, upsert=True)