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)

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) -> None:
    _require_asyncpg("PostgresGraphBackend")
    self._pool = pool

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.

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_subtree async

delete_subtree(scope: TenantScope) -> int

Erase scope and every descendant scope in one statement.

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 in one statement.

    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.
    """
    subtree_sql, subtree_params = sql_subtree_condition(scope, "scope_path", 2)
    conn = await self._pool.acquire()
    try:
        count = await conn.fetchval(
            "WITH doomed AS ("
            "  DELETE FROM memory_nodes"
            f"  WHERE tenant_id=$1 AND {subtree_sql}"
            "  RETURNING id"
            "), swept_edges AS ("
            "  DELETE FROM memory_edges"
            "  WHERE tenant_id=$1 AND ("
            "    source IN (SELECT id FROM doomed)"
            "    OR target IN (SELECT id 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:
        await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        await conn.execute(NODES_DDL)
        await conn.execute(EDGES_DDL)
        # v8.0 online-safe upgrade: add scope_path to pre-existing tables.
        await conn.execute(NODES_SCOPE_PATH_MIGRATION)
        # 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:
            await conn.execute(idx)
    finally:
        await self._pool.release(conn)

query_edges async

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

Query edges directly from memory_edges table with pagination.

Supports optional relationship filter. All other keys in filters are ignored to avoid SQL injection risk.

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 = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges directly from memory_edges table with pagination.

    Supports optional ``relationship`` filter. All other keys in ``filters``
    are ignored to avoid SQL injection risk.
    """
    conditions: list[str] = ["tenant_id=$1"]
    params: list[Any] = [scope.tenant_id]

    relationship = (filters or {}).get("relationship")
    if relationship is not None:
        params.append(relationship)
        conditions.append(f"relationship=${len(params)}")

    params.append(limit)
    limit_ph = f"${len(params)}"
    params.append(offset)
    offset_ph = f"${len(params)}"

    sql = (
        f"SELECT * FROM memory_edges WHERE {' AND '.join(conditions)}"
        f" ORDER BY created_at DESC"
        f" LIMIT {limit_ph} OFFSET {offset_ph}"
    )

    conn = await self._pool.acquire()
    try:
        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]

Query nodes at scope or below it (the descendant read).

Identical to :meth:query_nodes but for the scope predicate: the delimiter-terminated descendant sweep in place of the ancestor exact-IN. Sharing the body is deliberate — the two reads must agree about layer, label, ordering and truncation, and the only thing that may differ between them is which direction of the hierarchy they mean.

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]:
    """Query nodes at ``scope`` or below it (the descendant read).

    Identical to :meth:`query_nodes` but for the scope predicate: the
    delimiter-terminated descendant sweep in place of the ancestor
    exact-IN. Sharing the body is deliberate — the two reads must agree
    about layer, label, ordering and truncation, and the only thing that
    may differ between them is which direction of the hierarchy they mean.
    """
    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).

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``).
    """
    edge.tenant_id = scope.tenant_id
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            """
            INSERT INTO memory_edges
                (id, source, target, relationship, tenant_id, weight,
                 created_at, uses, provenance, source_episodic_ids,
                 scope_path)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb, $11)
            ON CONFLICT (tenant_id, source, target, relationship)
            DO UPDATE SET weight = memory_edges.weight + 1
            RETURNING *
            """,
            str(edge.id), str(edge.source), str(edge.target),
            edge.relationship, scope.tenant_id, edge.weight, edge.created_at,
            edge.uses, edge.provenance, json.dumps(edge.source_episodic_ids),
            materialise_scope_path(scope),
        )
    finally:
        await self._pool.release(conn)

    if row is None:  # pragma: no cover — RETURNING always yields a row
        return edge
    return _row_to_edge(row)

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.

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.
    """
    props = row["properties"]
    if isinstance(props, str):
        props = json.loads(props)
    props = dict(props)
    spreading = int(props.pop("__spreading_access_count", 0) or 0)
    # v8.0: if the dedicated scope_path column is populated but the JSONB
    # bag lacks it (e.g. a row written pre-v8.0 then backfilled), lift the
    # column value so the scorer dual-read sees a single source of truth.
    col_scope_path = _row_get(row, SCOPE_PATH_KEY)
    if col_scope_path and not props.get(SCOPE_PATH_KEY):
        props[SCOPE_PATH_KEY] = col_scope_path
    embedding = _decode_embedding(_row_get(row, "embedding"))
    return 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"],
    )