Skip to content

symfonic.memory.backends

backends

Backend implementations for graph and vector storage.

The Postgres backends import asyncpg (and pgvector) at module level behind a try/except ImportError guard. Exporting them eagerly here made that guard run for every import of anything under symfonic.memory -- including import symfonic.memory itself -- so a deployment with no Postgres still asked the interpreter for asyncpg on every process start. That is the "disabled integrations are not imported or probed" clause of story s4-2, and rule IA-5 of the integration-adapter standard (T4.2.1).

They are therefore resolved through a module __getattr__ (PEP 562): from symfonic.memory.backends import PostgresGraphBackend still works and still returns the same class, but the import cost is paid by the caller that asks for it. The in-memory backends stay eager -- they cost nothing.

InMemoryGraphBackend 

InMemoryGraphBackend()

GraphBackend over dicts keyed by (tenant_id, node_id/edge_id).

Testing and local development only.

Source code in symfonic/memory/backends/in_memory.py 
43
44
45
46
def __init__(self) -> None:
    self._nodes: dict[str, dict[str, MemoryNode]] = defaultdict(dict)
    self._edges: dict[str, dict[str, MemoryEdge]] = defaultdict(dict)
    self._workset_cursors: dict[Any, Any] = {}

add_edge async 

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

Add an edge to the in-memory store.

Source code in symfonic/memory/backends/in_memory.py 
111
112
113
114
115
async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Add an edge to the in-memory store."""
    edge.tenant_id = scope.tenant_id
    self._edges[scope.tenant_id][str(edge.id)] = edge
    return edge

add_node async 

add_node(scope: TenantScope, node: MemoryNode) -> MemoryNode

Add a node to the in-memory store.

v8.0: stamps the materialised scope_path into the node's property bag (zero-migration) so the prefix-isolation filter (§5.d) can be enforced on every read.

Source code in symfonic/memory/backends/in_memory.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
    """Add a node to the in-memory store.

    v8.0: stamps the materialised ``scope_path`` into the node's
    property bag (zero-migration) so the prefix-isolation filter (§5.d)
    can be enforced on every read.
    """
    node.tenant_id = scope.tenant_id
    node.properties[SCOPE_PATH_KEY] = materialise_scope_path(scope)
    self._nodes[scope.tenant_id][str(node.id)] = node
    return node

delete_edge async 

delete_edge(scope: TenantScope, edge_id: EdgeId) -> None

Delete a single edge by its ID.

Source code in symfonic/memory/backends/in_memory.py 
239
240
241
242
async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_edges.pop(str(edge_id), None)

delete_node async 

delete_node(scope: TenantScope, node_id: NodeId, cascade: bool = False) -> None

Delete a node, optionally cascading to connected edges.

Source code in symfonic/memory/backends/in_memory.py 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
async def delete_node(
    self, scope: TenantScope, node_id: NodeId, cascade: bool = False
) -> None:
    """Delete a node, optionally cascading to connected edges."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        return
    # v8.7.7: scope-gated by prefix visibility like the PG delete_node and
    # the get_node read filter, so a sibling sub-scope cannot delete this.
    stored = stored_scope_path(node.properties, scope.tenant_id)
    if not is_visible(stored, scope.ancestor_prefix_paths()):
        return
    tenant_nodes.pop(str(node_id), None)
    if cascade:
        tenant_edges = self._edges.get(scope.tenant_id, {})
        to_remove = [
            eid
            for eid, edge in tenant_edges.items()
            if str(edge.source) == str(node_id)
            or str(edge.target) == str(node_id)
        ]
        for eid in to_remove:
            del tenant_edges[eid]

delete_subtree async 

delete_subtree(scope: TenantScope) -> int

Erase owned subtree, edges and cursors in one non-yielding operation.

Source code in symfonic/memory/backends/in_memory.py 
244
245
246
247
async def delete_subtree(self, scope: TenantScope) -> int:
    """Erase owned subtree, edges and cursors in one non-yielding operation."""
    from symfonic.memory.backends.in_memory_deletion import delete_subtree
    return delete_subtree(self, scope)

get_neighbors async 

get_neighbors(scope: TenantScope, node_id: NodeId, relationship: str | None = None) -> list[MemoryNode]

Get nodes connected to the given node by edges.

Source code in symfonic/memory/backends/in_memory.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]:
    """Get nodes connected to the given node by edges."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    neighbor_ids: set[str] = set()
    for edge in tenant_edges.values():
        if relationship and edge.relationship != relationship:
            continue
        if str(edge.source) == str(node_id):
            neighbor_ids.add(str(edge.target))
        elif str(edge.target) == str(node_id):
            neighbor_ids.add(str(edge.source))
    return [
        tenant_nodes[nid]
        for nid in neighbor_ids
        if nid in tenant_nodes and self._node_visible(scope, tenant_nodes[nid])
    ]

get_node async 

get_node(scope: TenantScope, node_id: NodeId) -> MemoryNode | None

Get a node by ID within the scope's visible prefix chain.

Source code in symfonic/memory/backends/in_memory.py 
117
118
119
120
121
122
123
124
async def get_node(
    self, scope: TenantScope, node_id: NodeId
) -> MemoryNode | None:
    """Get a node by ID within the scope's visible prefix chain."""
    node = self._nodes.get(scope.tenant_id, {}).get(str(node_id))
    if node is None or not self._node_visible(scope, node):
        return None
    return node

query_edges async 

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

Query edges with optional relationship filter and pagination.

Source code in symfonic/memory/backends/in_memory.py 
264
265
266
267
268
269
270
271
272
273
async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges with optional relationship filter and pagination."""
    from symfonic.memory.backends.in_memory_queries import query_edges
    return query_edges(self._edges.get(scope.tenant_id, {}).values(), filters, limit, offset)

query_nodes async 

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

Filter nodes by properties within the tenant scope.

limit=None returns every matching node (no truncation); limit=0 returns no rows (v8.7.1 — unified across backends).

Source code in symfonic/memory/backends/in_memory.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
async def query_nodes(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = 50
) -> list[MemoryNode]:
    """Filter nodes by properties within the tenant scope.

    ``limit=None`` returns every matching node (no truncation);
    ``limit=0`` returns no rows (v8.7.1 — unified across backends).
    """
    if limit == 0:
        return []
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    results: list[MemoryNode] = []
    for node in tenant_nodes.values():
        if not self._node_visible(scope, node):
            continue
        if self._matches_filters(node, filters):
            results.append(node)
            if limit is not None and len(results) >= limit:
                break
    return results

query_subtree async 

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

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

Source code in symfonic/memory/backends/in_memory.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def query_subtree(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = None
) -> list[MemoryNode]:
    """Filter nodes at ``scope`` or below it (the descendant read)."""
    if limit == 0:
        return []
    root = materialise_scope_path(scope)
    results: list[MemoryNode] = []
    for node in self._nodes.get(scope.tenant_id, {}).values():
        stored = stored_scope_path(node.properties, node.tenant_id)
        if not is_in_subtree(stored, root):
            continue
        if self._matches_filters(node, filters):
            results.append(node)
            if limit is not None and len(results) >= limit:
                break
    return results

transaction 

transaction() -> Any

All of a batch, or none of it -- see :mod:.in_memory_transaction.

Source code in symfonic/memory/backends/in_memory.py 
38
39
40
41
def transaction(self) -> Any:
    """All of a batch, or none of it -- see :mod:`.in_memory_transaction`."""
    from symfonic.memory.backends.in_memory_transaction import snapshot
    return snapshot(self)

traverse async 

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

BFS traversal from start node up to max_depth.

Source code in symfonic/memory/backends/in_memory.py 
149
150
151
152
153
154
async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS traversal from start node up to max_depth."""
    from symfonic.memory.backends.in_memory_queries import traverse
    return await traverse(self, scope, start, max_depth)

update_node async 

update_node(scope: TenantScope, node_id: NodeId, updates: dict[str, Any]) -> MemoryNode

Partial update of node properties.

Source code in symfonic/memory/backends/in_memory.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
async def update_node(
    self, scope: TenantScope, node_id: NodeId, updates: dict[str, Any]
) -> MemoryNode:
    """Partial update of node properties."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        raise KeyError(f"Node {node_id} not found for tenant {scope.tenant_id}")
    for key, value in updates.items():
        if key == "updated_at" and isinstance(value, str):
            value = datetime.fromisoformat(value)
        if hasattr(node, key):
            object.__setattr__(node, key, value)
        else:
            node.properties[key] = value
    if "updated_at" not in updates:
        node.updated_at = datetime.now(UTC)
    return node

upsert_edge async 

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

Insert edge or increment weight by 1 if (tenant, source, target, rel) matches.

Source code in symfonic/memory/backends/in_memory.py 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if (tenant, source, target, rel) matches."""
    edge.tenant_id = scope.tenant_id
    tenant_edges = self._edges[scope.tenant_id]
    for existing in tenant_edges.values():
        if (
            str(existing.source) == str(edge.source)
            and str(existing.target) == str(edge.target)
            and existing.relationship == edge.relationship
        ):
            object.__setattr__(existing, "weight", existing.weight + 1)
            return existing
    tenant_edges[str(edge.id)] = edge
    return edge

InMemoryVectorBackend 

InMemoryVectorBackend()

In-memory implementation of the VectorBackend protocol.

Stores embeddings in a list and uses brute-force cosine similarity for search. Suitable for testing only.

Source code in symfonic/memory/backends/in_memory_vector.py 
49
50
def __init__(self) -> None:
    self._store: dict[str, list[dict[str, Any]]] = defaultdict(list)

add async 

add(scope: TenantScope, ids: list[str], embeddings: list[list[float]], metadatas: list[dict[str, Any]], documents: list[str]) -> None

Add vectors with metadata to the store.

v8.0: stamps the materialised scope_path into the metadata bag so the prefix-isolation filter (§5.d) can run on every search.

Source code in symfonic/memory/backends/in_memory_vector.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def add(
    self,
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None:
    """Add vectors with metadata to the store.

    v8.0: stamps the materialised ``scope_path`` into the metadata bag so
    the prefix-isolation filter (§5.d) can run on every search.
    """
    scope_path = materialise_scope_path(scope)
    for i, doc_id in enumerate(ids):
        meta = dict(metadatas[i]) if i < len(metadatas) else {}
        meta.setdefault(SCOPE_PATH_KEY, scope_path)
        self._store[scope.tenant_id].append({
            "id": doc_id,
            "embedding": embeddings[i],
            "metadata": meta,
            "document": documents[i] if i < len(documents) else "",
        })

count async 

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

Source code in symfonic/memory/backends/in_memory_vector.py 
122
123
124
async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    return len(self._store.get(scope.tenant_id, []))

delete async 

delete(scope: TenantScope, ids: list[str]) -> None

Delete vectors by their IDs.

Source code in symfonic/memory/backends/in_memory_vector.py 
105
106
107
108
109
110
111
async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    id_set = set(ids)
    if scope.tenant_id in self._store:
        self._store[scope.tenant_id] = [
            e for e in self._store[scope.tenant_id] if e["id"] not in id_set
        ]

search async 

search(scope: TenantScope, query_embedding: list[float], top_k: int = 5) -> list[dict[str, Any]]

Brute-force cosine similarity search.

v8.0: ALWAYS-ON prefix-isolation (§5.d) — only entries whose stored scope_path is a prefix of the query path are scored. Dual-reads pre-v8.0 metadata (no scope_path) as a 1-level root path.

Source code in symfonic/memory/backends/in_memory_vector.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
async def search(
    self,
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]:
    """Brute-force cosine similarity search.

    v8.0: ALWAYS-ON prefix-isolation (§5.d) — only entries whose stored
    ``scope_path`` is a prefix of the query path are scored.  Dual-reads
    pre-v8.0 metadata (no ``scope_path``) as a 1-level root path.
    """
    entries = self._store.get(scope.tenant_id, [])
    prefixes = scope.ancestor_prefix_paths()
    scored = []
    for entry in entries:
        entry_path = metadata_scope_path(entry["metadata"], scope.tenant_id)
        if not is_visible(entry_path, prefixes):
            continue
        sim = _cosine_similarity(query_embedding, entry["embedding"])
        scored.append({
            "id": entry["id"],
            "score": sim,
            "metadata": entry["metadata"],
            "document": entry["document"],
        })
    scored.sort(key=lambda x: x["score"], reverse=True)
    return scored[:top_k]

PostgresGraphBackend 

PostgresGraphBackend(pool: PostgresPoolManager)

GraphBackend backed by PostgreSQL with JSONB storage.

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

Source code in symfonic/memory/backends/postgres_graph.py 
355
356
357
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 symfonic/memory/backends/postgres_graph.py 
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
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 symfonic/memory/backends/postgres_graph.py 
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
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 symfonic/memory/backends/postgres_graph.py 
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
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 symfonic/memory/backends/postgres_graph.py 
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
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 symfonic/memory/backends/postgres_graph.py 
623
624
625
626
627
628
629
630
631
632
633
634
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 symfonic/memory/backends/postgres_graph.py 
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
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 symfonic/memory/backends/postgres_graph.py 
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
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 symfonic/memory/backends/postgres_graph.py 
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
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)

PostgresLeases 

PostgresLeases(pool: Any)

A :class:~symfonic.capabilities.memory.leases.LeasePort over Postgres.

Source code in symfonic/memory/backends/postgres_leases.py 
112
113
def __init__(self, pool: Any) -> None:
    self._pool = pool

pool property 

pool: Any

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

Public because the commit path has to prove that the lease, the graph and the staged records all live in one domain before it will call a cycle atomic. Reading that off a private attribute would make the proof depend on an implementation detail of this class.

ensure_schema async 

ensure_schema() -> None

Create the lease table if it is not there.

IF NOT EXISTS is idempotent across time and not across concurrency: two sessions running it at the same instant both pass the existence check and both insert into pg_type, and one gets a duplicate-key error. Ordinarily an obscure wrinkle -- here it is the expected case, because the workers this table exists to coordinate are exactly the ones that start together.

So a failure is retried once against a table that exists by then, and only then re-raised. Caught by the first run of the cross-process test and by nothing else: every later run found the table already there.

Source code in symfonic/memory/backends/postgres_leases.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
async def ensure_schema(self) -> None:
    """Create the lease table if it is not there.

    ``IF NOT EXISTS`` is idempotent across time and *not* across
    concurrency: two sessions running it at the same instant both pass the
    existence check and both insert into ``pg_type``, and one gets a
    duplicate-key error. Ordinarily an obscure wrinkle -- here it is the
    expected case, because the workers this table exists to coordinate are
    exactly the ones that start together.

    So a failure is retried once against a table that exists by then, and
    only then re-raised. Caught by the first run of the cross-process test
    and by nothing else: every later run found the table already there.
    """
    for attempt in (1, 2):
        conn = await self._pool.acquire()
        try:
            await conn.execute(LEASES_DDL)
            return
        except Exception:
            if attempt == 2:
                raise
            # Somebody else is creating it right now. The retry either
            # finds it there or fails for a reason that is not a race.
            logger.debug(
                "lease table creation raced another worker; retrying once",
                exc_info=True,
            )
        finally:
            await self._pool.release(conn)

hold_for_update async 

hold_for_update(lease: Lease) -> bool

Whether lease is ours, and keep it ours until this transaction ends.

The commit-time fence. :meth:holds answers a question about the past tense the moment it returns; this one takes the row lock, so a rival acquisition queues behind the caller's transaction instead of landing between the answer and the write it authorised.

Only meaningful inside a transaction -- outside one the lock is released with the statement and this is :meth:holds with extra cost. The commit path always calls it inside one; see :func:symfonic.capabilities.memory.commit.commit_cycle, which refuses to run without a transaction domain rather than calling this and pretending.

Source code in symfonic/memory/backends/postgres_leases.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
async def hold_for_update(self, lease: Lease) -> bool:
    """Whether ``lease`` is ours, *and* keep it ours until this transaction ends.

    The commit-time fence. :meth:`holds` answers a question about the past
    tense the moment it returns; this one takes the row lock, so a rival
    acquisition queues behind the caller's transaction instead of landing
    between the answer and the write it authorised.

    Only meaningful inside a transaction -- outside one the lock is released
    with the statement and this is :meth:`holds` with extra cost. The commit
    path always calls it inside one; see
    :func:`symfonic.capabilities.memory.commit.commit_cycle`, which refuses
    to run without a transaction domain rather than calling this and
    pretending.
    """
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchval(
            _HOLDS_FOR_UPDATE, lease.scope_path, lease.owner
        )
    finally:
        await self._pool.release(conn)
    return row is not None

PostgresPoolManager 

PostgresPoolManager(dsn: str, min_size: int = 2, max_size: int = 10)

Async connection pool manager wrapping asyncpg.Pool.

Usage::

pool = PostgresPoolManager(dsn="postgresql://user:pw@host/db")
await pool.open()
async with pool as conn:
    await conn.execute("SELECT 1")
await pool.close()

Or as a context manager for the pool itself::

async with PostgresPoolManager(dsn=...) as pool:
    conn = await pool.acquire()
    try:
        await conn.execute("SELECT 1")
    finally:
        await pool.release(conn)
Source code in symfonic/memory/backends/pool.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    dsn: str,
    min_size: int = 2,
    max_size: int = 10,
) -> None:
    if not _ASYNCPG_AVAILABLE:
        raise ImportError(
            "PostgresPoolManager requires asyncpg. "
            "Install it with: pip install symfonic-core[postgres]"
        )
    self._dsn = dsn
    self._min_size = min_size
    self._max_size = max_size
    self._pool: Any = None  # asyncpg.Pool at runtime

dsn property 

dsn: str

Return the configured connection string (read-only).

Used by the v7.1.1 PostgresCheckpointerFactory to build a separate psycopg pool against the same database. The two pools intentionally do not share connections (asyncpg vs. psycopg protocols are incompatible inside langgraph-checkpoint-postgres).

raw_pool property 

raw_pool: Any

Expose the underlying asyncpg.Pool for direct use (advanced).

acquire async 

acquire() -> Any

Acquire a connection from the pool.

Caller is responsible for releasing via :meth:release.

Inside :meth:transaction this hands back that transaction's own connection, so every backend on this pool joins it without knowing it exists (:mod:.pool_transaction).

Source code in symfonic/memory/backends/pool.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def acquire(self) -> Any:
    """Acquire a connection from the pool.

    Caller is responsible for releasing via :meth:`release`.

    Inside :meth:`transaction` this hands back that transaction's own
    connection, so every backend on this pool joins it without knowing it
    exists (:mod:`.pool_transaction`).
    """
    bound = bound_connection(self)
    if bound is not None:
        return bound
    if self._pool is None:
        raise RuntimeError(
            "Pool is not open. Call await pool.open() first "
            "or use 'async with pool' context manager."
        )
    return await self._pool.acquire()

close async 

close() -> None

Gracefully close all connections in the pool.

Source code in symfonic/memory/backends/pool.py 
106
107
108
109
110
async def close(self) -> None:
    """Gracefully close all connections in the pool."""
    if self._pool is not None:
        await self._pool.close()
        self._pool = None

open async 

open() -> None

Create the underlying asyncpg connection pool.

Passes init=_register_pgvector_on_connection so every new physical connection registers the pgvector binary codec before the pool hands it out. See module docstring for the rationale.

Source code in symfonic/memory/backends/pool.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def open(self) -> None:
    """Create the underlying asyncpg connection pool.

    Passes ``init=_register_pgvector_on_connection`` so every new
    physical connection registers the pgvector binary codec before
    the pool hands it out. See module docstring for the rationale.
    """
    if self._pool is not None:
        return
    self._pool = await asyncpg.create_pool(
        self._dsn,
        min_size=self._min_size,
        max_size=self._max_size,
        init=_register_pgvector_on_connection,
    )

release async 

release(conn: Any) -> None

Return a connection to the pool.

A no-op for a connection :meth:transaction holds: releasing it mid-way would hand a half-written batch to somebody else.

Source code in symfonic/memory/backends/pool.py 
135
136
137
138
139
140
141
142
143
144
async def release(self, conn: Any) -> None:
    """Return a connection to the pool.

    A no-op for a connection :meth:`transaction` holds: releasing it
    mid-way would hand a half-written batch to somebody else.
    """
    if is_bound(conn):
        return
    if self._pool is not None:
        await self._pool.release(conn)

transaction 

transaction() -> Any

Run a block so every operation on this pool commits or fails together.

Ambient rather than passed: :mod:.pool_transaction has the why.

Source code in symfonic/memory/backends/pool.py 
146
147
148
149
150
151
def transaction(self) -> Any:
    """Run a block so every operation on this pool commits or fails together.

    Ambient rather than passed: :mod:`.pool_transaction` has the why.
    """
    return run_in_transaction(self)

PostgresVectorBackend 

PostgresVectorBackend(pool: PostgresPoolManager, embedding_dim: int = 1536, *, search_strategy: Literal['exact', 'approximate'] = 'exact')

VectorBackend backed by PostgreSQL + pgvector.

Uses cosine distance (<=> operator) for similarity search. All operations enforce tenant isolation via WHERE tenant_id = $N. search_strategy='exact' ranks the authorized subset: bounded returned rows, but O(visible vectors) database distance work. 'approximate' opts into HNSW, which can underfill or miss neighbours after scope filtering.

Source code in symfonic/memory/backends/postgres_vector.py 
126
127
128
129
130
131
132
133
134
135
def __init__(self, pool: PostgresPoolManager, embedding_dim: int = 1536, *,
             search_strategy: Literal["exact", "approximate"] = "exact") -> None:
    _require_asyncpg("PostgresVectorBackend")
    if embedding_dim <= 0:
        raise ValueError("embedding_dim must be a positive integer")
    if search_strategy not in ("exact", "approximate"):
        raise ValueError("search_strategy must be exact or approximate")
    self._pool = pool
    self._embedding_dim = embedding_dim
    self.search_strategy = search_strategy

pool property 

pool: PostgresPoolManager

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

count async 

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

Source code in symfonic/memory/backends/postgres_vector.py 
279
280
281
282
283
284
285
286
287
288
289
async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            "SELECT COUNT(*)::int AS n FROM memory_vectors WHERE tenant_id=$1",
            scope.tenant_id,
        )
        return int(row["n"]) if row else 0
    finally:
        await self._pool.release(conn)

ensure_schema async 

ensure_schema() -> None

Create the memory_vectors table and indexes if not present.

Source code in symfonic/memory/backends/postgres_vector.py 
142
143
144
145
146
147
148
149
150
151
152
153
async def ensure_schema(self) -> None:
    """Create the memory_vectors table and indexes if not present."""
    conn = await self._pool.acquire()
    try:
        await conn.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        await conn.execute(VECTORS_DDL)
        await conn.execute(VECTORS_SCOPE_PATH_MIGRATION)
        for idx in VECTOR_INDEXES_DDL:
            await conn.execute(idx)
        await conn.execute(_ann_index_ddl(self._embedding_dim))
    finally:
        await self._pool.release(conn)