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.

EraseStep dataclass 

EraseStep(job_id: str | None, complete: bool, discarded: tuple[str, ...] = ())

What one eraser transaction left: complete, or a job still erasing.

EraserAttestation dataclass 

EraserAttestation(runtime_login: str, eraser_login: str, binding: Any)

What attest proved, once, on the eraser's own login (CORE-194).

InMemoryGraphBackend 

InMemoryGraphBackend()

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

Source code in src/symfonic/memory/backends/in_memory.py 
44
45
46
47
48
49
50
51
52
53
54
def __init__(self) -> None:
    from asyncio import Lock
    self._write_lock = Lock()
    self._nodes: dict[str, dict[str, MemoryNode]] = defaultdict(dict)
    self._edges: dict[str, dict[str, MemoryEdge]] = defaultdict(dict)
    self._workset_cursors: dict[Any, Any] = {}
    self._erased_subtrees: dict[str, set[str]] = defaultdict(set)
    self._privacy_erased_subtrees: dict[str, set[str]] = defaultdict(set)
    self._privacy_erased_tenants: set[str] = set()
    self._erase_epochs: dict[str, int] = defaultdict(int)
    self._positional_quarantine: dict[str, dict[Any, Any]] = defaultdict(dict)

add_edge async 

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

Add an edge; native endpoints must already be live and visible.

Source code in src/symfonic/memory/backends/in_memory.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
@serialized_write
async def add_edge(self, scope: StorageScope, edge: MemoryEdge) -> MemoryEdge:
    """Add an edge; native endpoints must already be live and visible."""
    from symfonic.memory.backends.in_memory_ownership import check_edge_endpoints
    check_edge_endpoints(self, scope, edge)
    path = materialise_scope_path(scope)
    existing = self._edges[scope.tenant_id].get(str(edge.id))
    if existing is not None and getattr(existing, "_scope_path", None) != path:
        raise ValueError("Edge ID already belongs to a different scope")
    edge.tenant_id = scope.tenant_id
    edge._scope_path = path
    self._edges[scope.tenant_id][str(edge.id)] = edge
    return edge

add_node async 

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

Stamp ownership; native writes cannot replace another owner.

Source code in src/symfonic/memory/backends/in_memory.py 
114
115
116
117
118
@serialized_write
async def add_node(self, scope: StorageScope, node: MemoryNode) -> MemoryNode:
    """Stamp ownership; native writes cannot replace another owner."""
    from symfonic.memory.backends.in_memory_ownership import replace_node
    return replace_node(self, scope, node)

delete_edge async 

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

Delete an edge visible on this scope's exact ancestor chain.

Source code in src/symfonic/memory/backends/in_memory.py 
260
261
262
263
264
265
266
267
268
269
@serialized_write
async def delete_edge(self, scope: StorageScope, edge_id: EdgeId) -> None:
    """Delete an edge visible on this scope's exact ancestor chain."""
    self._erase_epochs[scope.tenant_id] += 1
    tenant_edges = self._edges.get(scope.tenant_id, {})
    edge = tenant_edges.get(str(edge_id))
    if edge is not None and is_visible(
        getattr(edge, "_scope_path", None), scope.ancestor_prefix_paths()
    ):
        tenant_edges.pop(str(edge_id), None)

delete_node async 

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

Delete a node, optionally cascading to connected edges.

Source code in src/symfonic/memory/backends/in_memory.py 
253
254
255
256
257
258
259
@serialized_write
async def delete_node(
    self, scope: StorageScope, node_id: NodeId, cascade: bool = False,
) -> None:
    """Delete a node, optionally cascading to connected edges."""
    from symfonic.memory.backends.in_memory_deletion import delete_node
    delete_node(self, scope, node_id, cascade)

delete_subtree async 

delete_subtree(scope: StorageScope) -> int

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

Source code in src/symfonic/memory/backends/in_memory.py 
270
271
272
273
274
@serialized_write
async def delete_subtree(self, scope: StorageScope) -> 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: StorageScope, node_id: NodeId, relationship: str | None = None) -> list[MemoryNode]

Get nodes connected to the given node by edges.

Source code in src/symfonic/memory/backends/in_memory.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def get_neighbors(
    self,
    scope: StorageScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]:
    """Get nodes connected to the given node by edges."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    neighbor_ids: set[str] = set()
    from symfonic.memory.backends.in_memory_ownership import read_edges
    native = _is_native_scope(scope)
    if native and await self.get_node(scope, node_id) is None:
        return []
    for edge in read_edges(self, scope):
        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].model_copy(deep=True)
        for nid in neighbor_ids
        if nid in tenant_nodes and self._node_visible(scope, tenant_nodes[nid])
    ]

get_node async 

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

Get a node by ID on its visible ancestor chain.

Source code in src/symfonic/memory/backends/in_memory.py 
139
140
141
142
async def get_node(self, scope: StorageScope, node_id: NodeId) -> MemoryNode | None:
    """Get a node by ID on its visible ancestor chain."""
    node = self._nodes.get(scope.tenant_id, {}).get(str(node_id))
    return node.model_copy(deep=True) if node and self._node_visible(scope, node) else None

patch_owned_node async 

patch_owned_node(scope: StorageScope, node_id: NodeId, expected: OwnedNodeVersion, properties: dict[str, Any]) -> MemoryNode | None

Merge properties only if exact owner, incarnation and revision match.

Source code in src/symfonic/memory/backends/in_memory.py 
217
218
219
220
221
222
223
async def patch_owned_node(
    self, scope: StorageScope, node_id: NodeId, expected: OwnedNodeVersion,
    properties: dict[str, Any],
) -> MemoryNode | None:
    """Merge properties only if exact owner, incarnation and revision match."""
    from symfonic.memory.backends.in_memory_owned_node_patch import patch
    return await patch(self, scope, node_id, expected, properties)

query_edges async 

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

Query tenant edges, optionally restricted to induced endpoints.

Source code in src/symfonic/memory/backends/in_memory.py 
290
291
292
293
294
295
296
297
298
299
300
async def query_edges(
    self,
    scope: StorageScope,
    filters: dict[str, Any] | None = None,
    limit: int | None = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query tenant edges, optionally restricted to induced endpoints."""
    from symfonic.memory.backends.in_memory_ownership import read_edges
    from symfonic.memory.backends.in_memory_queries import query_edges
    return query_edges(read_edges(self, scope), filters, limit, offset, scope)

query_nodes async 

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

Filter visible nodes; None is unlimited and zero returns no rows.

Source code in src/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_nodes(
    self, scope: StorageScope, filters: dict[str, Any], limit: int | None = 50
) -> list[MemoryNode]:
    """Filter visible nodes; None is unlimited and zero returns no rows."""
    if limit == 0:
        return []
    from symfonic.memory.backends.in_memory_queries import matches_filters
    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 matches_filters(node, filters):
            results.append(node.model_copy(deep=True))
            if limit is not None and len(results) >= limit:
                break
    return results

query_subtree async 

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

Filter owned nodes at or below scope.

Source code in src/symfonic/memory/backends/in_memory.py 
195
196
197
198
199
200
async def query_subtree(
    self, scope: StorageScope, filters: dict[str, Any], limit: int | None = None
) -> list[MemoryNode]:
    """Filter owned nodes at or below scope."""
    from symfonic.memory.backends.in_memory_queries import query_subtree
    return query_subtree(self, scope, filters, limit)

read_owned_node async 

read_owned_node(scope: StorageScope, node_id: NodeId) -> tuple[MemoryNode, OwnedNodeVersion] | None

Return a detached exact-owner row with its immutable stored guard.

Source code in src/symfonic/memory/backends/in_memory.py 
210
211
212
213
214
215
async def read_owned_node(
    self, scope: StorageScope, node_id: NodeId,
) -> tuple[MemoryNode, OwnedNodeVersion] | None:
    """Return a detached exact-owner row with its immutable stored guard."""
    from symfonic.memory.backends.in_memory_owned_node_patch import read
    return await read(self, scope, node_id)

transaction 

transaction() -> Any

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

Source code in src/symfonic/memory/backends/in_memory.py 
39
40
41
42
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: StorageScope, start: NodeId, max_depth: int) -> list[MemoryNode]

BFS traversal from start node up to max_depth.

Source code in src/symfonic/memory/backends/in_memory.py 
170
171
172
173
174
175
async def traverse(
    self, scope: StorageScope, 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: StorageScope, node_id: NodeId, updates: dict[str, Any]) -> MemoryNode

Partial update of node properties.

Source code in src/symfonic/memory/backends/in_memory.py 
202
203
204
205
206
207
208
@serialized_write
async def update_node(
    self, scope: StorageScope, node_id: NodeId, updates: dict[str, Any]
) -> MemoryNode:
    """Partial update of node properties."""
    from symfonic.memory.backends.in_memory_node_update import update
    return update(self, scope, node_id, updates)

upsert_edge async 

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

Insert or increment an edge matching scope, endpoints, and relation.

Source code in src/symfonic/memory/backends/in_memory.py 
284
285
286
287
288
@serialized_write
async def upsert_edge(self, scope: StorageScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert or increment an edge matching scope, endpoints, and relation."""
    from symfonic.memory.backends.in_memory_owned_edge import upsert
    return upsert(self, scope, 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 src/symfonic/memory/backends/in_memory_vector.py 
55
56
def __init__(self) -> None:
    self._store: dict[str, list[dict[str, Any]]] = defaultdict(list)

add async 

add(scope: StorageScope, 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 src/symfonic/memory/backends/in_memory_vector.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 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 add(
    self,
    scope: StorageScope,
    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)
    from symfonic.memory.backends.in_memory_ownership import native_owner
    native = _is_native_scope(scope)
    wanted = set(ids)
    for entry in self._store.get(scope.tenant_id, []):
        owner = metadata_scope_path(entry["metadata"], scope.tenant_id, scope=scope)
        if entry["id"] in wanted and (
            owner != scope_path if native else native_owner(owner)
        ):
            from symfonic.memory.errors import VectorOwnershipConflict
            raise VectorOwnershipConflict("Vector ID already belongs to a different scope")
    if not native and any(native_owner(meta.get(SCOPE_PATH_KEY)) for meta in metadatas):
        raise ValueError("Legacy metadata cannot declare a native scope owner")
    added = []
    for i, doc_id in enumerate(ids):
        meta = dict(metadatas[i]) if i < len(metadatas) else {}
        if native:
            meta[SCOPE_PATH_KEY] = scope_path
        else:
            meta.setdefault(SCOPE_PATH_KEY, scope_path)
        entries = added if native else self._store[scope.tenant_id]
        entries.append({
            "id": doc_id,
            "embedding": embeddings[i],
            "metadata": meta,
            "document": documents[i] if i < len(documents) else "",
        })
    if native:
        replacements = {row["id"]: row for row in added}
        self._store[scope.tenant_id] = [
            row for row in self._store.get(scope.tenant_id, ())
            if row["id"] not in replacements
        ] + list(replacements.values())

add_owned_batch async 

add_owned_batch(entries: list[tuple[Any, str, list[float], dict, str]]) -> None

Validate every native owner before changing vectors; never awaits.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
async def add_owned_batch(self, entries: list[tuple[Any, str, list[float], dict, str]]) -> None:
    """Validate every native owner before changing vectors; never awaits."""
    from symfonic.memory.errors import VectorOwnershipConflict
    proposed: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for scope, identifier, embedding, metadata, document in entries:
        if not _is_native_scope(scope):
            raise ValueError("owned vector batch requires native scopes")
        path = materialise_scope_path(scope)
        for row in (*self._store.get(scope.tenant_id, ()), *proposed[scope.tenant_id]):
            if row["id"] == identifier and metadata_scope_path(
                row["metadata"], scope.tenant_id, scope=scope,
            ) != path:
                raise VectorOwnershipConflict("Vector ID already belongs to a different scope")
        proposed[scope.tenant_id].append({
            "id": identifier, "embedding": embedding,
            "metadata": {**metadata, SCOPE_PATH_KEY: path}, "document": document,
        })
    for tenant, rows in proposed.items():
        replacements = {row["id"]: row for row in rows}
        self._store[tenant] = [
            row for row in self._store.get(tenant, ())
            if row["id"] not in replacements
        ] + list(replacements.values())

count async 

count(scope: StorageScope) -> int

Return the total number of stored vectors for the tenant.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
193
194
195
196
197
198
199
200
201
202
203
async def count(self, scope: StorageScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    entries = self._store.get(scope.tenant_id, [])
    if not _is_native_scope(scope):
        from symfonic.memory.backends.in_memory_ownership import native_owner
        return sum(not native_owner(entry["metadata"].get(SCOPE_PATH_KEY))
                   for entry in entries)
    return sum(is_visible(
        metadata_scope_path(entry["metadata"], scope.tenant_id, scope=scope),
        scope.ancestor_prefix_paths(),
    ) for entry in entries)

delete async 

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

Delete vectors by their IDs.

Source code in src/symfonic/memory/backends/in_memory_vector.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
async def delete(self, scope: StorageScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    from symfonic.memory.backends.in_memory_ownership import native_owner
    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
            or (not _is_native_scope(scope) and native_owner(
                e["metadata"].get(SCOPE_PATH_KEY),
            ))
            or (_is_native_scope(scope) and not is_visible(
                metadata_scope_path(e["metadata"], scope.tenant_id, scope=scope),
                scope.ancestor_prefix_paths(),
            ))
        ]

search async 

search(scope: StorageScope, query_embedding: list[float], top_k: int = 5, *, allowed_scope_paths: tuple[str, ...] | None = None) -> 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 src/symfonic/memory/backends/in_memory_vector.py 
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
156
157
158
159
160
161
162
163
164
async def search(
    self,
    scope: StorageScope,
    query_embedding: list[float],
    top_k: int = 5,
    *,
    allowed_scope_paths: tuple[str, ...] | None = None,
) -> 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()
    from symfonic.memory.backends.in_memory_ownership import native_owner
    scored = []
    native = _is_native_scope(scope)
    for entry in entries:
        entry_path = metadata_scope_path(entry["metadata"], scope.tenant_id, scope=scope)
        if not native and native_owner(entry_path):
            continue
        if (native or allowed_scope_paths is None) and not is_visible(entry_path, prefixes):
            continue
        if allowed_scope_paths is not None and entry_path not in allowed_scope_paths:
            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]

LegacyEdgeBootstrapRefused

Bases: RuntimeError

A default host cannot restore the tenant-wide index after cutover.

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 
374
375
376
377
378
379
380
381
382
383
384
385
386
387
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 
405
406
407
408
409
410
411
412
413
414
415
416
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 
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
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 
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
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 
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
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
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 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
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 
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
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 
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
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 
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
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 
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
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 
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
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)

PostgresGraphRuntime 

PostgresGraphRuntime(pool: Any, *, owner_login: str, edge_upsert_mode: str = 'tenant_guarded', eraser: Any = None)

Bases: PostgresGraphBackend

Explicit nonowner graph adapter; ensure_schema performs read-only checks.

Source code in src/symfonic/memory/backends/postgres_graph_runtime.py 
168
169
170
171
172
173
174
175
176
177
def __init__(
    self,
    pool: Any,
    *,
    owner_login: str,
    edge_upsert_mode: str = "tenant_guarded",
    eraser: Any = None,
) -> None:
    super().__init__(pool, edge_upsert_mode=edge_upsert_mode, eraser=eraser)
    self._owner_login = owner_login

attest_eraser async 

attest_eraser() -> Any

The runtime role contract, then the eraser's proof for this runtime.

One runtime connection proves the role contract and resolves the session_user and memory_nodes the eraser must match (CORE-194). Read-only. Any refusal withdraws the standing attestation.

Source code in src/symfonic/memory/backends/postgres_graph_runtime.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
async def attest_eraser(self) -> Any:
    """The runtime role contract, then the eraser's proof for this runtime.

    One runtime connection proves the role contract and resolves the
    ``session_user`` and ``memory_nodes`` the eraser must match
    (CORE-194). Read-only. Any refusal withdraws the standing attestation.
    """
    from symfonic.memory.backends.postgres_native_eraser import runtime_identity

    if self.eraser is None:
        raise MemoryContractError("no native eraser is composed (CORE-199)")
    self._runtime_identity = None
    connection = await self.pool.acquire()
    try:
        await require_runtime(connection, self._edge_upsert_mode, self._owner_login)
        identity = await runtime_identity(_Pinned(connection))
    finally:
        await self.pool.release(connection)
    attestation = await self.eraser.attest(runtime_login=identity[0], binding=identity[1])
    self._runtime_identity = identity
    return attestation

ensure_schema async 

ensure_schema() -> None

Read-only startup; with an eraser composed, also attest_eraser.

Source code in src/symfonic/memory/backends/postgres_graph_runtime.py 
179
180
181
182
183
184
185
186
187
188
async def ensure_schema(self) -> None:
    """Read-only startup; with an eraser composed, also ``attest_eraser``."""
    if self.eraser is not None:
        await self.attest_eraser()
        return
    connection = await self.pool.acquire()
    try:
        await require_runtime(connection, self._edge_upsert_mode, self._owner_login)
    finally:
        await self.pool.release(connection)

PostgresLeases 

PostgresLeases(pool: Any)

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

Source code in src/symfonic/memory/backends/postgres_leases.py 
126
127
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 table and native index under a schema-scoped DDL lock.

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 src/symfonic/memory/backends/postgres_leases.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
async def ensure_schema(self) -> None:
    """Create table and native index under a schema-scoped DDL lock.

    ``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:
            async with conn.transaction():
                # Functions and expression indexes race just like tables.
                await conn.execute(
                    "SELECT pg_advisory_xact_lock(2147483646, hashtext(current_schema()))"
                )
                await conn.execute(LEASES_DDL)
                await conn.execute(NATIVE_INDEX_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 src/symfonic/memory/backends/postgres_leases.py 
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
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.
    """
    scope = scope_from_path(lease.scope_path)
    conn = await self._pool.acquire()
    try:
        # Acquisition of a child row also needs to wait for this commit.
        await conn.execute(_LOCK_TENANT, scope.tenant)
        row = await conn.fetchval(
            _HOLDS_FOR_UPDATE, lease.scope_path, lease.owner
        )
    finally:
        await self._pool.release(conn)
    return row is not None

PostgresNativeEraser 

PostgresNativeEraser(pool: Any, *, schema: str, receipt_key: bytes, erase_anchor: Any = None, receipt_key_id: str | None = None, receipt_verification_keys: Any = None, anchor_key: bytes | None = None)

Job creation, chunking and completion on the eraser's own login.

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    pool: Any,
    *,
    schema: str,
    receipt_key: bytes,
    erase_anchor: Any = None,
    receipt_key_id: str | None = None,
    receipt_verification_keys: Any = None,
    anchor_key: bytes | None = None,
) -> None:
    from symfonic.memory.backends.postgres_retirement_receipt import configure

    configure(
        self,
        pool,
        schema,
        receipt_key,
        erase_anchor,
        receipt_key_id,
        receipt_verification_keys,
        anchor_key,
    )
    self._attested: dict[tuple[str, Any], EraserAttestation] = {}

attest async 

attest(*, runtime_login: str, binding: Any) -> EraserAttestation

D6 on this login, binding's very graph, and not runtime_login.

Read-only. Refuses with EraserNotReady and withdraws every earlier attestation, so a failed re-attestation cannot leave a stale proof.

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
 98
 99
100
101
102
103
104
105
106
107
async def attest(self, *, runtime_login: str, binding: Any) -> EraserAttestation:
    """D6 on this login, ``binding``'s very graph, and not ``runtime_login``.

    Read-only. Refuses with ``EraserNotReady`` and withdraws every earlier
    attestation, so a failed re-attestation cannot leave a stale proof.
    """
    login = await self._probe(runtime_login, binding)
    attestation = EraserAttestation(runtime_login, login, binding)
    self._attested[(runtime_login, binding)] = attestation
    return attestation

attestation_for 

attestation_for(runtime_login: str, binding: Any) -> EraserAttestation | None

The standing attestation for that runtime identity, or None.

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
138
139
140
def attestation_for(self, runtime_login: str, binding: Any) -> EraserAttestation | None:
    """The standing attestation for that runtime identity, or None."""
    return self._attested.get((runtime_login, binding))

guarding 

guarding(runtime_pool: Any) -> PostgresNativeEraser

Guard the runtime's pool with this eraser's anchor proof (CORE-171).

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
83
84
85
86
87
88
def guarding(self, runtime_pool: Any) -> PostgresNativeEraser:
    """Guard the runtime's pool with this eraser's anchor proof (CORE-171)."""
    from symfonic.memory.backends.postgres_erase_incarnation import install

    install(runtime_pool, self._guard)
    return self

require_ready async 

require_ready() -> None

Startup: refuse unless this login holds exactly D6 in its schema.

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
94
95
96
async def require_ready(self) -> None:
    """Startup: refuse unless this login holds exactly D6 in its schema."""
    await self._probe(None, None)

resume async 

resume(job_id: str) -> EraseStep

Erase one chunk of job_id in the configured schema; the operator's entry.

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
191
192
193
194
195
196
async def resume(self, job_id: str) -> EraseStep:
    """Erase one chunk of ``job_id`` in the configured schema; the operator's entry."""
    if type(job_id) is not str or not job_id:
        raise MemoryContractError("native erase job id must be a nonempty string")
    await self._verified()
    return await self._advance(job_id, None)

retire_tenant async 

retire_tenant(scope: Any, *, runtime_login: str, binding: Any) -> EraseStep

Explicit permanent tenant retirement; ordinary start stays reusable.

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
146
147
148
149
150
async def retire_tenant(self, scope: Any, *, runtime_login: str, binding: Any) -> EraseStep:
    """Explicit permanent tenant retirement; ordinary ``start`` stays reusable."""
    from symfonic.memory.backends.postgres_erase_retirement import retire

    return await retire(self, scope, runtime_login=runtime_login, binding=binding)

start async 

start(scope: Any, *, runtime_login: str, binding: Any, vectors: bool = False) -> EraseStep

Erase content and create the job, or resume the identical one.

Source code in src/symfonic/memory/backends/postgres_native_eraser.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
async def start(
    self, scope: Any, *, runtime_login: str, binding: Any, vectors: bool = False
) -> EraseStep:
    """Erase content and create the job, or resume the identical one."""
    from symfonic.memory.backends.postgres_erase_create import active, create
    from symfonic.memory.backends.postgres_erase_routing import (
        OVERLAPPING,
        overlap_args,
        overlap_sql,
    )
    from symfonic.memory.backends.postgres_native_subtree import GRAPH_LOCK

    await self._verified()
    async with self._transaction(binding) as conn:
        jobs = await self._jobs(conn, runtime_login)
        resumed = await active(conn, jobs, scope)
        if resumed is None:
            await conn.execute(_LIFECYCLE_LOCK, f"symfonic-memory:{scope.tenant_id}")
            await conn.execute(GRAPH_LOCK)
            # A start of this same scope may have committed its job while
            # this one waited; that job is resumed, not an overlap.
            resumed = await active(conn, jobs, scope)
        if resumed is None:
            if await conn.fetch(overlap_sql(jobs), *overlap_args(scope)):
                raise MemoryContractError(OVERLAPPING)
            if vectors:
                from symfonic.memory.backends.postgres_erase_vector_subtree import preflight

                await preflight(conn, jobs, scope)
            created, chunked = await create(conn, jobs, scope, vectors=vectors)
            if not chunked:
                return created
            resumed = created.job_id
        elif vectors:
            from symfonic.memory.backends.postgres_erase_vector_subtree import require_job

            await require_job(conn, jobs, resumed)
    return await self._advance(resumed, binding)

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 src/symfonic/memory/backends/pool.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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
    self._guards: tuple[Any, ...] = ()  # CORE-171 erase anchor: postgres_erase_incarnation

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(*, timeout: float | None = None) -> 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 src/symfonic/memory/backends/pool.py 
121
122
123
124
125
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
async def acquire(self, *, timeout: float | None = None) -> 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`).
    """
    if foreign_bound_transaction(self):
        raise RuntimeError("inherited PostgreSQL transaction belongs to another task")
    bound = bound_connection(self)
    if bound is not None:
        for guard in self._guards:
            await guard(bound)
        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."
        )
    if timeout is None:
        conn = await self._pool.acquire()
    else:
        conn = await self._pool.acquire(timeout=timeout)
    if self._guards:
        try:
            for guard in self._guards:  # every checkout, before any caller
                await guard(conn)
        except BaseException:
            await self._pool.release(conn)
            raise
    return conn

anchor_graph 

anchor_graph(eraser: Any) -> Any

CORE-171: guard this pool with eraser's anchor proof, or refuse native reads.

Source code in src/symfonic/memory/backends/pool.py 
155
156
157
158
159
def anchor_graph(self, eraser: Any) -> Any:
    """CORE-171: guard this pool with ``eraser``'s anchor proof, or refuse native reads."""
    from symfonic.memory.backends.postgres_erase_incarnation import anchored_graph

    return anchored_graph(self, eraser)

close async 

close() -> None

Gracefully close all connections in the pool.

Source code in src/symfonic/memory/backends/pool.py 
111
112
113
114
115
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

guard_connections 

guard_connections(guard: Any) -> None

Also run guard on every connection acquire hands out, before the caller.

Guards only add: a connection is handed out once every guard admitted it. A placeholder guard (refuse: nothing proves this pool) yields to, and is never added beside, a guard that proves every checkout.

Source code in src/symfonic/memory/backends/pool.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def guard_connections(self, guard: Any) -> None:
    """Also run ``guard`` on every connection ``acquire`` hands out, before the caller.

    Guards only add: a connection is handed out once every guard admitted it.
    A ``placeholder`` guard (refuse: nothing proves this pool) yields to,
    and is never added beside, a guard that proves every checkout.
    """
    domain = getattr(guard, "guard_domain", "native")
    same_domain = tuple(
        g for g in self._guards if getattr(g, "guard_domain", "native") == domain
    )
    proving = tuple(g for g in same_domain if not getattr(g, "placeholder", False))
    if getattr(guard, "placeholder", False):
        if not proving and guard not in same_domain:
            self._guards += (guard,)
    elif guard not in proving:
        # One authority cannot substitute for an unrelated authority.
        self._guards = tuple(
            g
            for g in self._guards
            if getattr(g, "guard_domain", "native") != domain
            or not getattr(g, "placeholder", False)
        ) + (guard,)

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 src/symfonic/memory/backends/pool.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
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, *, timeout: float | None = None) -> 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 src/symfonic/memory/backends/pool.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
async def release(self, conn: Any, *, timeout: float | None = None) -> 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:
        if timeout is None:
            await self._pool.release(conn)
            return
        task = asyncio.ensure_future(self._pool.release(conn, timeout=max(0, timeout)))
        try:
            done, _ = await asyncio.wait({task}, timeout=max(0, timeout))
            if not done:
                raise TimeoutError("PostgreSQL release was not acknowledged")
            task.result()
        except BaseException:
            # A detached proxy may already have been released; never mask cancellation.
            with suppress(Exception):
                conn.terminate()  # Discard is not evidence of transaction rollback.
            task.cancel()
            task.add_done_callback(
                lambda finished: None if finished.cancelled() else finished.exception()
            )
            raise

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 src/symfonic/memory/backends/pool.py 
213
214
215
216
217
218
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 src/symfonic/memory/backends/postgres_vector.py 
110
111
112
113
114
115
116
117
118
119
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

Count tenant vectors subject to permanent retirement visibility.

Source code in src/symfonic/memory/backends/postgres_vector.py 
287
288
289
290
291
292
293
294
295
296
297
298
299
300
async def count(self, scope: TenantScope) -> int:
    """Count tenant vectors subject to permanent retirement visibility."""
    from symfonic.memory.backends.postgres_vector_retirement import reader

    conn = await self._pool.acquire()
    try:
        row = await conn.fetchrow(
            "SELECT COUNT(*)::int AS n FROM memory_vectors WHERE tenant_id=$1 "
            f"AND {await reader(conn)}",
            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 src/symfonic/memory/backends/postgres_vector.py 
126
127
128
129
130
131
132
133
134
135
136
137
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)

RuntimeRoleNotReady

Bases: RuntimeError

A login violates the migrator/runtime split contract.

grant_runtime async 

grant_runtime(connection: Any, runtime_login: str, *, schema: str = 'public', ledger: str = MIGRATION_LEDGER) -> tuple[str, ...]

Owner side: reset the runtime to exact DML on every owned table, atomically.

Every table in schema gets SELECT/INSERT/UPDATE/DELETE and nothing else, column grants included; the CORE-199 erase job table gets SELECT only, its receipts nothing, its memory erase generations SELECT/INSERT/UPDATE and its erase log SELECT/INSERT; the migration ledger, views, materialized views and foreign tables get nothing; sequences get USAGE only. No default privileges are installed, so each migration must be followed by this call. Returns the tables granted DML.

Source code in src/symfonic/memory/backends/postgres_runtime_roles.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
async def grant_runtime(
    connection: Any, runtime_login: str, *, schema: str = "public", ledger: str = MIGRATION_LEDGER
) -> tuple[str, ...]:
    """Owner side: reset the runtime to exact DML on every owned table, atomically.

    Every table in ``schema`` gets SELECT/INSERT/UPDATE/DELETE and nothing else,
    column grants included; the CORE-199 erase job table gets SELECT only,
    its receipts nothing, its memory erase generations SELECT/INSERT/UPDATE and its
    erase log SELECT/INSERT; the
    migration ledger, views, materialized views and foreign tables get nothing;
    sequences get USAGE only. No default privileges are installed, so each
    migration must be followed by this call. Returns the tables granted DML.
    """
    runtime, quoted_schema = _identifier(runtime_login), _identifier(schema)
    async with connection.transaction():
        if await connection.fetchval("SELECT current_user=$1", runtime_login):
            raise RuntimeRoleNotReady("the owner cannot grant to itself")
        await _unprivileged(connection, runtime_login)
        rows = await _owned_schema(connection, schema)
        await connection.execute(f"REVOKE CREATE ON SCHEMA {quoted_schema} FROM {runtime}")
        await connection.execute(f"GRANT USAGE ON SCHEMA {quoted_schema} TO {runtime}")
        granted = []
        for row in rows:
            # Revoking a table's privileges also revokes its column privileges.
            await connection.execute(f"REVOKE ALL ON {_target(schema, row)} FROM {runtime}")
            if row["relkind"] == "S":
                await connection.execute(f"GRANT USAGE ON {_target(schema, row)} TO {runtime}")
            elif row["relkind"] in _TABLE_KINDS and row["relname"] in _READ_ONLY:
                if privileges := ", ".join(_READ_ONLY[row["relname"]]):
                    await connection.execute(
                        f"GRANT {privileges} ON {_target(schema, row)} TO {runtime}"
                    )
            elif row["relkind"] in _TABLE_KINDS and row["relname"] != ledger:
                await connection.execute(
                    f"GRANT {', '.join(_DML)} ON {_target(schema, row)} TO {runtime}"
                )
                granted.append(row["relname"])
        await _unprivileged(connection, runtime_login)
    return tuple(granted)

prove_terminal async 

prove_terminal(eraser: Any, tenant_id: str, job_id: str, *, runtime_login: str, binding: Any) -> TerminalRetirementEvidence

Recheck one exact, versioned, empty tenant-root retirement under its fence.

Do not call retire_tenant here: a proof request must never initiate an irreversible retirement. This is a read-side prerequisite, not an issuer.

Source code in src/symfonic/memory/backends/postgres_retirement_evidence.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
async def prove_terminal(
    eraser: Any, tenant_id: str, job_id: str, *, runtime_login: str, binding: Any
) -> TerminalRetirementEvidence:
    """Recheck one exact, versioned, empty tenant-root retirement under its fence.

    Do not call ``retire_tenant`` here: a proof request must never initiate an
    irreversible retirement. This is a read-side prerequisite, not an issuer.
    """
    from symfonic.capabilities.memory.scope import MemoryScope
    from symfonic.memory.backends.postgres_erase_content import content_ready
    from symfonic.memory.backends.postgres_erase_retirement import installed, require_empty
    from symfonic.memory.backends.postgres_native_eraser import PostgresNativeEraser
    from symfonic.memory.backends.postgres_retirement_receipt import initiation, verify
    from symfonic.memory.storage_scope import NativeStorageScope

    if (
        type(eraser) is not PostgresNativeEraser
        or type(tenant_id) is not str
        or not tenant_id
        or type(job_id) is not str
        or not job_id
    ):
        raise MemoryContractError(INVALID)
    root = NativeStorageScope.from_scope(MemoryScope.from_levels((("tenant", tenant_id),)))
    await eraser._verified()
    async with eraser._transaction(binding) as conn:
        jobs = await eraser._jobs(conn, runtime_login)
        if not await installed(conn, jobs) or not await content_ready(conn, jobs):
            raise MemoryContractError(INVALID)
        if eraser._receipt_keys is None:
            raise MemoryContractError(INVALID)
        await initiation(conn, jobs, eraser)
        row = await conn.fetchrow(
            f"SELECT tenant_id, prefix, receipt_key_id, receipt_binding, receipt_anchor "
            f"FROM {jobs} WHERE id=$1 AND tenant_id=$2 AND permanent "
            "AND state='retired' FOR SHARE",
            job_id,
            tenant_id,
        )
        if row is None or row["prefix"] != root.subtree_prefix or not row["receipt_key_id"]:
            raise MemoryContractError(INVALID)
        await require_empty(conn, jobs, job_id)
        receipts = jobs.rsplit(".", 1)[0] + ".memory_erase_receipts"
        await verify(conn, jobs, receipts, job_id, eraser)
        anchor_head = await eraser.proof.verify(conn, pinned=True)
        return TerminalRetirementEvidence(
            tenant_id,
            job_id,
            row["receipt_binding"],
            row["receipt_anchor"],
            row["receipt_key_id"],
            anchor_head,
        )

require_legacy_bootstrap_safe async 

require_legacy_bootstrap_safe(conn: Any) -> None

Refuse default index DDL once the scoped index has replaced it.

The resolved table is the same unqualified table used by ensure_schema. A fresh database has neither index and keeps legacy bootstrap behavior.

Source code in src/symfonic/memory/backends/postgres_edge_index_preflight.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
async def require_legacy_bootstrap_safe(conn: Any) -> None:
    """Refuse default index DDL once the scoped index has replaced it.

    The resolved table is the same unqualified table used by ensure_schema.
    A fresh database has neither index and keeps legacy bootstrap behavior.
    """
    try:
        indexes = await conn.fetchrow(_BOOTSTRAP_SQL, OLD_INDEX, NEW_INDEX)
        if indexes is None or any(
            indexes[key] is not True and indexes[key] is not False
            for key in ("old_present", "new_present", "old_usable")
        ):
            raise LegacyEdgeBootstrapRefused("edge bootstrap catalog is unavailable")
        if indexes["new_present"] and not indexes["old_usable"]:
            raise LegacyEdgeBootstrapRefused(
                "scoped edge index is active; default edge bootstrap is refused"
            )
    except LegacyEdgeBootstrapRefused:
        raise
    except Exception:
        raise LegacyEdgeBootstrapRefused("edge bootstrap catalog check failed") from None

require_scoped_edge_index async 

require_scoped_edge_index(conn: Any) -> None

Check the exact table resolved by the writer on this connection.

Source code in src/symfonic/memory/backends/postgres_edge_scoped_writer.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def require_scoped_edge_index(conn: Any) -> None:
    """Check the exact table resolved by the writer on this connection."""
    try:
        indexes = await conn.fetch(_INDEX_SQL)
        if (
            not indexes or indexes[0]["relkind"] != "r"
            or indexes[0]["identity_collations_deterministic"] is not True
        ):
            raise ScopedEdgeIndexNotReady("scoped edge index is not ready")
        for index in indexes:
            keys = index["keys"]
            if index["name"] == OLD_INDEX or (
                index["unique"] and index["key_count"] == 4
                and tuple(keys) == _KEYS
            ):
                raise ScopedEdgeIndexNotReady("tenant-wide edge index is still present")
            if (
                index["name"] != NEW_INDEX and index["unique"]
                and index["key_count"] == 5 and tuple(keys[:4]) == _KEYS
            ):
                # ON CONFLICT may infer every matching unique index. An extra
                # expression index can arbitrate with a different collation.
                raise ScopedEdgeIndexNotReady("another scoped edge arbiter is present")
        matches = [index for index in indexes if index["name"] == NEW_INDEX]
        if len(matches) != 1:
            raise ScopedEdgeIndexNotReady("scoped edge index is not ready")
        index = matches[0]
        keys = index["keys"]
        if not (
            index["unique"] and index["valid"] and index["ready"]
            and index["live"] and index["nonpartial"]
            and index["key_collations_deterministic"] is True
            and index["key_opclasses_default"] is True
            and index["method"] == "btree"
            and index["key_count"] == index["total_count"] == 5
            and len(keys) == 5 and tuple(keys[:4]) == _KEYS
            and _canonical_expression(keys[4]) in _EXPRESSIONS
        ):
            raise ScopedEdgeIndexNotReady("scoped edge index is not ready")
    except ScopedEdgeIndexNotReady:
        raise
    except Exception:
        # Driver errors can contain SQL values. Keep refusal independent of them.
        raise ScopedEdgeIndexNotReady("scoped edge index verification failed") from None

require_split_runtime async 

require_split_runtime(connection: Any, *, owner_login: str, tables: tuple[str, ...], schema: str = 'public', ledger: str = MIGRATION_LEDGER) -> None

Runtime side, read-only: refuse startup unless this login is DML-only.

The connected login must be a direct, unprivileged, membership-free LOGIN owning nothing (large objects included), able to create nothing, holding no large object grant and unable to EXECUTE any SECURITY DEFINER function outside pg_catalog and information_schema. schema and every relation in it must belong to owner_login; each table carries exactly the table-level DML grant and no column grant beyond it (the ledger none, the erase job table SELECT only, its receipts none, the memory erase generations SELECT/INSERT/UPDATE, the erase log SELECT/INSERT), every tables entry and memory_nodes resolve on this login's search_path to schema, views, materialized views and foreign tables carry none, every tables entry exists, and sequences are USAGE without UPDATE.

Source code in src/symfonic/memory/backends/postgres_runtime_roles.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
async def require_split_runtime(
    connection: Any,
    *,
    owner_login: str,
    tables: tuple[str, ...],
    schema: str = "public",
    ledger: str = MIGRATION_LEDGER,
) -> None:
    """Runtime side, read-only: refuse startup unless this login is DML-only.

    The connected login must be a direct, unprivileged, membership-free LOGIN
    owning nothing (large objects included), able to create nothing, holding no
    large object grant and unable to EXECUTE any SECURITY DEFINER function
    outside ``pg_catalog`` and ``information_schema``. ``schema`` and every
    relation in it must belong to ``owner_login``; each table carries exactly
    the table-level DML grant and no column grant beyond it (the ledger none,
    the erase job table SELECT only, its receipts none, the memory erase
    generations SELECT/INSERT/UPDATE, the erase log SELECT/INSERT), every ``tables`` entry
    and ``memory_nodes`` resolve on this login's search_path to ``schema``,
    views, materialized views and foreign tables carry none, every ``tables``
    entry exists, and sequences are USAGE without UPDATE.
    """
    if not await connection.fetchval("SELECT session_user=current_user"):
        raise RuntimeRoleNotReady("runtime must not run under SET ROLE")
    login = await connection.fetchval("SELECT current_user")
    if login == owner_login:
        raise RuntimeRoleNotReady("runtime must not be the schema owner")
    await _unprivileged(connection, login)
    if not await connection.fetchval(
        "SELECT pg_get_userbyid(nspowner)=$2 FROM pg_namespace WHERE nspname=$1",
        schema,
        owner_login,
    ):
        raise RuntimeRoleNotReady("application schema must belong to the named owner")
    rows = await connection.fetch(_RELATIONS, schema)
    present = {row["relname"] for row in rows if row["relkind"] in _TABLE_KINDS}
    if not all(row["schema_owned"] for row in rows) or not set(tables) <= present:
        raise RuntimeRoleNotReady(
            "required tables must exist and every relation belong to the owner"
        )
    # The native erase fence reads jobs beside the memory_nodes this login
    # resolves; attesting another schema's grants would prove nothing there.
    for name in sorted({*tables, "memory_nodes"} & present):
        if await connection.fetchval(_RESOLVES_TO, name) != schema:
            raise RuntimeRoleNotReady(f"{name} must resolve to the attested schema {schema}")
    for held in await connection.fetch(_HELD, schema, list(_DML + _EXCESS)):
        table, privilege = held["relname"], held["privilege"]
        wanted = _READ_ONLY.get(table, _DML)
        if held["relkind"] in _TABLE_KINDS and table != ledger and privilege in wanted:
            ok = held["on_table"]
        else:
            ok = not held["anywhere"]
        if not ok:
            raise RuntimeRoleNotReady(
                f"runtime privilege {privilege} on {table} "
                f"must be {'absent' if held['anywhere'] else 'held'}"
            )
    for sequence in (row for row in rows if row["relkind"] == "S"):
        target = f'{_identifier(schema)}."{sequence["relname"]}"'
        if not await connection.fetchval(
            "SELECT has_sequence_privilege(current_user, $1, 'USAGE') AND NOT "
            "has_sequence_privilege(current_user, $1, 'UPDATE')",
            target,
        ):
            raise RuntimeRoleNotReady("runtime sequences must be USAGE without UPDATE")

seal 

seal(evidence: TerminalRetirementEvidence, chat: ChatRetirementIntent, *, witness_key_id: str, witness_key: bytes, receipt_key: bytes, anchor_key: bytes) -> SignedRetirementWitness

Seal already-verified evidence; do not use from the API/worker process.

Source code in src/symfonic/memory/backends/postgres_retirement_witness.py 
 71
 72
 73
 74
 75
 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
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def seal(
    evidence: TerminalRetirementEvidence,
    chat: ChatRetirementIntent,
    *,
    witness_key_id: str,
    witness_key: bytes,
    receipt_key: bytes,
    anchor_key: bytes,
) -> SignedRetirementWitness:
    """Seal already-verified evidence; do not use from the API/worker process."""
    if type(evidence) is not TerminalRetirementEvidence or type(chat) is not ChatRetirementIntent:
        raise MemoryContractError(INVALID)
    witness_key = _key(witness_key)
    if hmac.compare_digest(witness_key, _key(receipt_key)) or hmac.compare_digest(
        witness_key, _key(anchor_key)
    ):
        raise MemoryContractError("retirement witness key must differ from receipt and anchor keys")
    if (
        evidence.tenant_id != chat.tenant_id
        or not all(
            _required(x)
            for x in (
                evidence.tenant_id,
                evidence.job_id,
                evidence.graph_binding,
                evidence.graph_anchor,
                evidence.receipt_key_id,
                chat.chat_db_identity,
                chat.chat_anchor_digest,
                witness_key_id,
            )
        )
        or type(chat.generation) is not int
        or chat.generation < 1
        or type(evidence.anchor_head) is not int
        or evidence.anchor_head < 0
        or type(chat.chat_anchor_seq) is not int
        or chat.chat_anchor_seq < 1
        or not _hex64(chat.manifest_sha256)
        or not _hex64(chat.chat_anchor_digest)
        or len(witness_key_id) > 128
        or len(evidence.receipt_key_id) > 128
    ):
        raise MemoryContractError(INVALID)
    claims = RetirementClaims(
        1,
        evidence.tenant_id,
        chat.generation,
        evidence.graph_binding,
        evidence.graph_anchor,
        chat.chat_db_identity,
        evidence.job_id,
        evidence.receipt_key_id,
        evidence.anchor_head,
        chat.chat_anchor_seq,
        chat.chat_anchor_digest,
        chat.manifest_sha256,
        witness_key_id,
    )
    encoded = json.dumps(asdict(claims), sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    mac = hmac.new(witness_key, DOMAIN + encoded.encode(), hashlib.sha256).hexdigest()
    return SignedRetirementWitness(encoded, mac)