symfonic.memory.backends.in_memory

in_memory

In-memory backends for testing and prototyping.

InMemoryGraphBackend

InMemoryGraphBackend()

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

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

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

@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

@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

@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

@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

@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

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

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

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

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

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

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

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

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

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

@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

@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

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

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

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

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

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

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]