symfonic.memory.backends.postgres_graph

postgres_graph

PostgreSQL graph backend using JSONB storage.

Implements GraphBackend protocol with asyncpg for all database I/O. All operations enforce tenant isolation via WHERE tenant_id = $N.

asyncpg must be installed.

pip install symfonic-core[postgres]

PostgresGraphBackend

PostgresGraphBackend(pool: PostgresPoolManager, *, edge_upsert_mode: str = 'tenant_guarded', eraser: Any = None)

Bases: PostgresReviewCas

GraphBackend backed by PostgreSQL with JSONB storage.

All operations enforce tenant isolation via WHERE tenant_id = $N.

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

def __init__(
    self, pool: PostgresPoolManager, *, edge_upsert_mode: str = "tenant_guarded",
    eraser: Any = None,
) -> None:
    _require_asyncpg("PostgresGraphBackend")
    if edge_upsert_mode not in {"tenant_guarded", "scoped"}:
        raise ValueError("unknown edge upsert mode")
    if eraser is not None and eraser.pool is pool:
        raise ValueError("the native eraser needs its own pool and login (CORE-199 D6)")
    self._pool = pool
    self._edge_upsert_mode = edge_upsert_mode
    from symfonic.memory.backends.postgres_erase_incarnation import graph_eraser
    self._eraser = graph_eraser(pool, eraser)  # CORE-171
    self._runtime_identity: tuple[str, Any] | None = None

eraser property

eraser: Any

The explicitly composed CORE-199 eraser principal, or None.

eraser_attestation property

eraser_attestation: Any

The eraser's standing proof for this runtime, or None (CORE-194).

Read-only: present only after :meth:attest_eraser succeeded and no later attestation or eraser transaction has seen it contradicted.

pool property

pool: PostgresPoolManager

The pool this backend speaks through, and so its transaction domain.

Public because the consolidation commit has to prove that the graph, the lease and the staged records share one domain before it will call a cycle atomic. A proof that read a private attribute would be a proof about this class's spelling.

attest_eraser async

attest_eraser() -> Any

Prove the eraser holds D6, erases this runtime's very memory_nodes and logs in as a different session_user; read-only, else refuse.

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

async def attest_eraser(self) -> Any:
    """Prove the eraser holds D6, erases this runtime's very ``memory_nodes``
    and logs in as a different ``session_user``; read-only, else refuse."""
    from symfonic.capabilities.memory.errors import MemoryContractError
    from symfonic.memory.backends.postgres_native_eraser import runtime_identity

    if self._eraser is None:
        raise MemoryContractError("no native eraser is composed (CORE-199)")
    identity = await runtime_identity(self._pool)
    attestation = await self._eraser.attest(runtime_login=identity[0], binding=identity[1])
    self._runtime_identity = identity
    return attestation

backfill_scope_paths async

backfill_scope_paths() -> int

Eagerly backfill NULL scope_path rows (design §6.b migration).

Pre-v8.0 rows have scope_path IS NULL; they read as a 1-level root path tenant\x1f<tenant_id> via the query-time dual-read, so this eager backfill is OPTIONAL (online-safe, idempotent) — it just materialises the same value into the column so the B-tree index and the scorer's dual-read agree. Returns the number of rows updated. Uses a parameterised expression so adopter tenant_ids cannot inject.

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

async def backfill_scope_paths(self) -> int:
    """Eagerly backfill NULL ``scope_path`` rows (design §6.b migration).

    Pre-v8.0 rows have ``scope_path IS NULL``; they read as a 1-level root
    path ``tenant\\x1f<tenant_id>`` via the query-time dual-read, so this
    eager backfill is OPTIONAL (online-safe, idempotent) — it just
    materialises the same value into the column so the B-tree index and
    the scorer's dual-read agree.  Returns the number of rows updated.
    Uses a parameterised expression so adopter tenant_ids cannot inject.
    """
    # The 1-level root path is f"tenant\x1f{tenant_id}"; build it in SQL.
    delim = "\x1f"
    conn = await self._pool.acquire()
    try:
        result = await conn.execute(
            "UPDATE memory_nodes "
            "SET scope_path = 'tenant' || $1 || tenant_id "
            "WHERE scope_path IS NULL",
            delim,
        )
    finally:
        await self._pool.release(conn)
    # asyncpg returns e.g. "UPDATE 42"
    try:
        return int(str(result).split()[-1])
    except (ValueError, IndexError):
        return 0

delete_native_subtree_ids async

delete_native_subtree_ids(scope: TenantScope) -> tuple[str, ...]

Return precisely the native nodes erased by the locked sweep.

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

async def delete_native_subtree_ids(self, scope: TenantScope) -> tuple[str, ...]:
    """Return precisely the native nodes erased by the locked sweep."""
    from symfonic.memory.backends.postgres_forget_deadline import forget_transaction
    from symfonic.memory.backends.postgres_native_subtree import delete_graph_ids
    from symfonic.memory.storage_scope import NativeStorageScope

    if not isinstance(scope, NativeStorageScope):
        raise TypeError("native subtree scope required")
    from symfonic.capabilities.memory.errors import MemoryContractError
    from symfonic.memory.backends.postgres_erase_routing import route_content, route_forget
    from symfonic.memory.backends.postgres_pending_claims import erase_subtree

    async with forget_transaction(self._pool) as conn:
        await require_direct_edge_index(conn, self._edge_upsert_mode)
        await route_forget(conn, scope)  # hands off to the eraser before any delete
        try:
            doomed = await delete_graph_ids(conn, scope)
        except MemoryContractError as error:  # refused before any delete
            await route_content(conn, error)
            raise
        await erase_subtree(conn, scope)
        return doomed

delete_subtree async

delete_subtree(scope: TenantScope) -> int

Erase scope and every descendant scope.

Native v1 validates persisted owners under a transaction and table locks.

For v0, a data-modifying CTE, not two round trips and not a client-side scan: one statement is one implicit transaction, so a write that lands in a descendant scope while this runs is either wholly before the sweep's snapshot (and erased) or wholly after it. The enumerate-then-delete shape has a window between those two facts, and the operation that opens it is a tenant deletion running while other work drains.

The predicate is the delimiter-terminated descendant sweep from :mod:symfonic.memory.subtree — the one place a LIKE on a scope path is correct, and only because the delimiter is appended. Without it, forgetting acme would take acmecorp with it.

Edges go if either endpoint went, or if the edge itself sits in the subtree; the same two placeholders serve both halves.

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

async def delete_subtree(self, scope: TenantScope) -> int:
    """Erase ``scope`` and every descendant scope.

    Native v1 validates persisted owners under a transaction and table locks.

    For v0, a data-modifying CTE, not two round trips and not a client-side scan:
    one statement is one implicit transaction, so a write that lands in a
    descendant scope while this runs is either wholly before the sweep's
    snapshot (and erased) or wholly after it. The enumerate-then-delete
    shape has a window between those two facts, and the operation that
    opens it is a tenant deletion running while other work drains.

    The predicate is the delimiter-terminated descendant sweep from
    :mod:`symfonic.memory.subtree` — the one place a ``LIKE`` on a scope
    path is correct, and only because the delimiter is appended. Without
    it, forgetting ``acme`` would take ``acmecorp`` with it.

    Edges go if either endpoint went, or if the edge itself sits in the
    subtree; the same two placeholders serve both halves.
    """
    from symfonic.memory.storage_scope import NativeStorageScope

    if isinstance(scope, NativeStorageScope):
        return len(await self.delete_native_subtree_ids(scope))

    subtree_sql, subtree_params = sql_subtree_condition(scope, "scope_path", 2)
    conn = await self._pool.acquire()
    try:
        await require_direct_edge_index(conn, self._edge_upsert_mode)
        count = await conn.fetchval(
            "WITH doomed AS ("
            "  DELETE FROM memory_nodes"
            f'  WHERE tenant_id=$1 AND tenant_id COLLATE "C"=$1 AND {subtree_sql}'
            "  RETURNING id"
            "), swept_edges AS ("
            "  DELETE FROM memory_edges"
            '  WHERE tenant_id=$1 AND tenant_id COLLATE "C"=$1 AND ('
            '    (source IN (SELECT id FROM doomed) AND '
            '     source COLLATE "C" IN (SELECT id COLLATE "C" FROM doomed))'
            '    OR (target IN (SELECT id FROM doomed) AND '
            '     target COLLATE "C" IN (SELECT id COLLATE "C" FROM doomed))'
            f"    OR {subtree_sql}"
            "  )"
            "), swept_progress AS ("
            f"  DELETE FROM memory_workset_progress WHERE {subtree_sql}"
            ") SELECT count(*) FROM doomed",
            scope.tenant_id, *subtree_params,
        )
    finally:
        await self._pool.release(conn)
    return int(count or 0)

ensure_schema async

ensure_schema() -> None

Create graph tables (nodes + edges) and their indexes.

Does NOT create the memory_vectors table — that is the responsibility of :class:PostgresVectorBackend.ensure_schema.

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

async def ensure_schema(self) -> None:
    """Create graph tables (nodes + edges) and their indexes.

    Does NOT create the ``memory_vectors`` table — that is the
    responsibility of :class:`PostgresVectorBackend.ensure_schema`.
    """
    conn = await self._pool.acquire()
    try:
        if self._edge_upsert_mode == "tenant_guarded":
            from symfonic.memory.backends.postgres_edge_index_preflight import (
                require_legacy_bootstrap_safe,
            )
            await require_legacy_bootstrap_safe(conn)
        await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        await conn.execute(NODES_DDL)
        await conn.execute(EDGES_DDL)
        from symfonic.memory.backends.postgres_pending_claims import ensure as ensure_claims
        await ensure_claims(conn)
        # v8.0 online-safe upgrade: add scope_path to pre-existing tables.
        await conn.execute(NODES_SCOPE_PATH_MIGRATION)
        from symfonic.memory.backends import postgres_node_version_schema

        await postgres_node_version_schema.ensure(conn)
        # Online-safe upgrade: add edge model fields to pre-existing tables.
        await conn.execute(EDGES_FIELDS_MIGRATION)
        from symfonic.memory.backends.postgres_worksets import ensure
        await ensure(conn)
        for idx in GRAPH_INDEXES_DDL:
            if self._edge_upsert_mode == "scoped" and "idx_memory_edges_upsert" in idx:
                continue
            if self._edge_upsert_mode == "tenant_guarded" and "idx_memory_edges_upsert" in idx:
                await require_legacy_bootstrap_safe(conn)
            await conn.execute(idx)
        if self._edge_upsert_mode == "scoped":
            from symfonic.memory.backends.postgres_edge_scoped_writer import (
                require_scoped_edge_index,
            )
            await require_scoped_edge_index(conn)
    finally:
        await self._pool.release(conn)
    if self._eraser is not None:  # CORE-199 D6 and CORE-194 binding: proven here
        await self.attest_eraser()

query_edges async

query_edges(scope: TenantScope, filters: dict[str, Any] | None = None, limit: int | None = 50, offset: int = 0) -> list[MemoryEdge]

Query edges directly from memory_edges table with pagination.

Supports relationship, induced/incident endpoints, and opt-in edge owner visibility for v0. Native edge/endpoint admission is mandatory before pagination. Other keys are ignored; all values are bound.

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

async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int | None = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges directly from memory_edges table with pagination.

    Supports relationship, induced/incident endpoints, and opt-in edge
    owner visibility for v0. Native edge/endpoint admission is mandatory
    before pagination. Other keys are ignored; all values are bound.
    """
    from symfonic.memory.backends.postgres_native_edge_reads import query_statement

    conn = await self._pool.acquire()
    try:
        from symfonic.memory.backends.postgres_working_expiry_reads import read_visibility
        statement = query_statement(scope, filters, limit, offset,
                                    await read_visibility(conn, scope))
        if statement is None:
            return []
        sql, params = statement
        rows = await conn.fetch(sql, *params)
    finally:
        await self._pool.release(conn)

    return [_row_to_edge(row) for row in rows]

query_subtree async

query_subtree(scope: TenantScope, filters: dict[str, Any], limit: int | None = None) -> list[MemoryNode]

Descendant read; native phases share a transaction and time budget.

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

async def query_subtree(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = None
) -> list[MemoryNode]:
    """Descendant read; native phases share a transaction and time budget."""
    from symfonic.memory.scope_isolation import _is_native_scope

    if _is_native_scope(scope):
        from symfonic.memory.backends.postgres_native_subtree_query import (
            NATIVE_SUBTREE_BUDGET,
            canonical_node_paths,
            exact_paths_condition,
            native_subtree_deadline,
        )

        conn = await self._pool.acquire()
        try:
            async with native_subtree_deadline(conn):
                paths = await canonical_node_paths(
                    conn, scope, filters.get("id"), max_paths=NATIVE_SUBTREE_BUDGET,
                )
                if not paths:
                    return []
                return await self._query_nodes(
                    scope, filters, limit,
                    lambda _scope, col, i: exact_paths_condition(paths, col, i),
                    hard_cap=NATIVE_SUBTREE_BUDGET, conn=conn,
                )
        finally:
            await self._pool.release(conn)
    return await self._query_nodes(scope, filters, limit, sql_subtree_condition)

query_subtree_page async

query_subtree_page(scope: TenantScope, filters: dict[str, Any], *, limit: int, offset: int) -> list[MemoryNode]

One native descendant page without reading every prior page.

GraphAdminService accepts offsets for its browser contract. Its compatibility fallback must fetch through offset + limit because the historical backend protocol has no offset; Postgres does, and at volume that fallback turns page fifty into a ten-thousand-row process allocation. Keep the optional seam on the concrete adapter until the backend protocol can add it without breaking adopters.

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

async def query_subtree_page(
    self,
    scope: TenantScope,
    filters: dict[str, Any],
    *,
    limit: int,
    offset: int,
) -> list[MemoryNode]:
    """One native descendant page without reading every prior page.

    ``GraphAdminService`` accepts offsets for its browser contract.  Its
    compatibility fallback must fetch through ``offset + limit`` because
    the historical backend protocol has no offset; Postgres does, and at
    volume that fallback turns page fifty into a ten-thousand-row process
    allocation.  Keep the optional seam on the concrete adapter until the
    backend protocol can add it without breaking adopters.
    """
    if limit <= 0:
        return []
    if offset < 0:
        raise ValueError("offset must be non-negative")
    return await self._query_nodes(
        scope,
        filters,
        limit,
        sql_subtree_condition,
        offset=offset,
    )

traverse async

traverse(scope: TenantScope, start: NodeId, max_depth: int) -> list[MemoryNode]

BFS graph traversal up to max_depth hops from start node.

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

async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS graph traversal up to max_depth hops from start node."""
    visited: set[str] = set()
    queue: deque[tuple[str, int]] = deque([(str(start), 0)])
    result: list[MemoryNode] = []
    while queue:
        current_id, depth = queue.popleft()
        if current_id in visited:
            continue
        visited.add(current_id)
        node = await self.get_node(scope, NodeId(current_id))
        if node is not None:
            result.append(node)
        if depth < max_depth:
            neighbors = await self.get_neighbors(scope, NodeId(current_id))
            for nb in neighbors:
                if str(nb.id) not in visited:
                    queue.append((str(nb.id), depth + 1))
    return result

upsert_edge async

upsert_edge(scope: TenantScope, edge: MemoryEdge) -> MemoryEdge

Insert edge or increment weight by 1 if the same edge already exists.

Uniqueness is determined by (tenant_id, source, target, relationship). The unique index idx_memory_edges_upsert must exist for ON CONFLICT to resolve correctly (created by ensure_schema). Until the index includes owner scope, a sibling collision is refused rather than incrementing another owner's edge. Legacy NULL paths belong to the tenant root.

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

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if the same edge already exists.

    Uniqueness is determined by (tenant_id, source, target, relationship).
    The unique index ``idx_memory_edges_upsert`` must exist for ON CONFLICT
    to resolve correctly (created by ``ensure_schema``). Until the index
    includes owner scope, a sibling collision is refused rather than
    incrementing another owner's edge. Legacy NULL paths belong to the
    tenant root.
    """
    from symfonic.memory.backends.postgres_native_edge_write import upsert_edge

    return await upsert_edge(self._pool, scope, edge, self._edge_upsert_mode)

matches_extra

matches_extra(node: MemoryNode, filters: dict[str, Any]) -> bool

Apply arbitrary property filters not handled by SQL.

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

def matches_extra(node: MemoryNode, filters: dict[str, Any]) -> bool:
    """Apply arbitrary property filters not handled by SQL."""
    for key, value in filters.items():
        if hasattr(node, key):
            if getattr(node, key) != value:
                return False
        elif key not in node.properties or node.properties[key] != value:
            return False
    return True

row_to_node

row_to_node(row: Any) -> MemoryNode

Convert an asyncpg Record to a MemoryNode.

v6.2 T02 note: spreading_access_count has no dedicated column (no schema migration was required by the v6.2 carve-out). It is round-tripped through the properties JSONB under the reserved key __spreading_access_count; on read we lift the field out of properties so the Python model remains clean, and on write _node_properties_jsonb serialises it back in.

v7.2.1 hotfix: the embedding pgvector column is now hydrated onto MemoryNode.embedding. Prior versions silently dropped the column on read, which collapsed RetrievalScorer.semantic_similarity (weight 0.4) to zero for every Postgres-loaded graph candidate.

The persisted graph guard is private model state but participates in model equality. Hydrate it so independent reads of one row share a preimage.

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

def row_to_node(row: Any) -> MemoryNode:
    """Convert an asyncpg Record to a MemoryNode.

    v6.2 T02 note: ``spreading_access_count`` has no dedicated column
    (no schema migration was required by the v6.2 carve-out). It is
    round-tripped through the ``properties`` JSONB under the reserved
    key ``__spreading_access_count``; on read we lift the field out of
    properties so the Python model remains clean, and on write
    ``_node_properties_jsonb`` serialises it back in.

    v7.2.1 hotfix: the ``embedding`` pgvector column is now hydrated
    onto ``MemoryNode.embedding``. Prior versions silently dropped the
    column on read, which collapsed ``RetrievalScorer.semantic_similarity``
    (weight 0.4) to zero for every Postgres-loaded graph candidate.

    The persisted graph guard is private model state but participates in model
    equality. Hydrate it so independent reads of one row share a preimage.
    """
    props = row["properties"]
    if isinstance(props, str):
        props = json.loads(props)
    props = dict(props)
    spreading = int(props.pop("__spreading_access_count", 0) or 0)
    # The column, not user JSONB, owns scope_path (including legacy NULL).
    col_scope_path = _row_get(row, SCOPE_PATH_KEY)
    from symfonic.memory.backends.node_ownership import owned_properties

    props = owned_properties(props, col_scope_path)
    embedding = _decode_embedding(_row_get(row, "embedding"))
    node = MemoryNode(
        id=NodeId(row["id"]),
        layer=MemoryLayer(row["layer"]),
        tenant_id=row["tenant_id"],
        label=row["label"],
        properties=props,
        embedding=embedding,
        created_at=row["created_at"],
        updated_at=row["updated_at"],
        access_count=row["access_count"],
        spreading_access_count=spreading,
        importance=row["importance"],
    )
    incarnation = _row_get(row, "graph_incarnation")
    if incarnation is not None:
        node._graph_incarnation = str(incarnation)
        node._graph_revision = row["graph_revision"]
    return node