symfonic.memory.backends.in_memory

in_memory

In-memory backends for testing and prototyping.

Zero-external-dependency implementations of GraphBackend and VectorBackend. All data is stored in plain Python dicts, scoped by tenant_id.

InMemoryGraphBackend

InMemoryGraphBackend()

In-memory implementation of the GraphBackend protocol.

Stores nodes and edges in dicts keyed by (tenant_id, node_id/edge_id). Suitable for testing and local development only.

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

def __init__(self) -> None:
    self._nodes: dict[str, dict[str, MemoryNode]] = defaultdict(dict)
    self._edges: dict[str, dict[str, MemoryEdge]] = defaultdict(dict)

add_edge async

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

Add an edge to the in-memory store.

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

async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Add an edge to the in-memory store."""
    edge.tenant_id = scope.tenant_id
    self._edges[scope.tenant_id][str(edge.id)] = edge
    return edge

add_node async

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

Add a node to the in-memory store.

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

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

async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
    """Add a node to the in-memory store.

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

delete_edge async

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

Delete a single edge by its ID.

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

async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_edges.pop(str(edge_id), None)

delete_node async

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

Delete a node, optionally cascading to connected edges.

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

async def delete_node(
    self, scope: TenantScope, node_id: NodeId, cascade: bool = False
) -> None:
    """Delete a node, optionally cascading to connected edges."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        return
    # v8.7.7: scope-gate deletes by prefix visibility so a sibling sub-scope
    # cannot delete another sub-scope's node (mirrors the PG delete_node and
    # get_node read filter; previously tenant-only). Round-2 fix.
    stored = stored_scope_path(node.properties, scope.tenant_id)
    if not is_visible(stored, scope.ancestor_prefix_paths()):
        return
    tenant_nodes.pop(str(node_id), None)
    if cascade:
        tenant_edges = self._edges.get(scope.tenant_id, {})
        to_remove = [
            eid
            for eid, edge in tenant_edges.items()
            if str(edge.source) == str(node_id)
            or str(edge.target) == str(node_id)
        ]
        for eid in to_remove:
            del tenant_edges[eid]

get_neighbors async

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

Get nodes connected to the given node by edges.

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

async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[MemoryNode]:
    """Get nodes connected to the given node by edges."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    neighbor_ids: set[str] = set()
    for edge in tenant_edges.values():
        if relationship and edge.relationship != relationship:
            continue
        if str(edge.source) == str(node_id):
            neighbor_ids.add(str(edge.target))
        elif str(edge.target) == str(node_id):
            neighbor_ids.add(str(edge.source))
    return [
        tenant_nodes[nid]
        for nid in neighbor_ids
        if nid in tenant_nodes and self._node_visible(scope, tenant_nodes[nid])
    ]

get_node async

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

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

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

async def get_node(
    self, scope: TenantScope, node_id: NodeId
) -> MemoryNode | None:
    """Get a node by ID within the scope's visible prefix chain."""
    node = self._nodes.get(scope.tenant_id, {}).get(str(node_id))
    if node is None or not self._node_visible(scope, node):
        return None
    return node

query_edges async

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

Query edges with optional relationship filter and pagination.

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

async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[MemoryEdge]:
    """Query edges with optional relationship filter and pagination."""
    tenant_edges = self._edges.get(scope.tenant_id, {})
    relationship = (filters or {}).get("relationship")

    results: list[MemoryEdge] = []
    for edge in tenant_edges.values():
        if relationship is not None and edge.relationship != relationship:
            continue
        results.append(edge)

    # Stable ordering: newest first (mirrors Postgres behaviour)
    results.sort(key=lambda e: e.created_at, reverse=True)
    return results[offset : offset + limit]

query_nodes async

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

Filter nodes by properties within the tenant scope.

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

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

async def query_nodes(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = 50
) -> list[MemoryNode]:
    """Filter nodes by properties within the tenant scope.

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

traverse async

traverse(
    scope: TenantScope, 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: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS traversal from start node up to max_depth."""
    visited: set[str] = set()
    queue: list[tuple[str, int]] = [(str(start), 0)]
    result: list[MemoryNode] = []
    tenant_nodes = self._nodes.get(scope.tenant_id, {})

    while queue:
        current_id, depth = queue.pop(0)
        if current_id in visited:
            continue
        visited.add(current_id)
        node = tenant_nodes.get(current_id)
        if node is not None and self._node_visible(scope, node):
            result.append(node)
        if depth < max_depth:
            neighbors = await self.get_neighbors(scope, NodeId(current_id))
            for neighbor in neighbors:
                if str(neighbor.id) not in visited:
                    queue.append((str(neighbor.id), depth + 1))
    return result

update_node async

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

Partial update of node properties.

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

async def update_node(
    self, scope: TenantScope, node_id: NodeId, updates: dict[str, Any]
) -> MemoryNode:
    """Partial update of node properties."""
    tenant_nodes = self._nodes.get(scope.tenant_id, {})
    node = tenant_nodes.get(str(node_id))
    if node is None:
        raise KeyError(f"Node {node_id} not found for tenant {scope.tenant_id}")
    for key, value in updates.items():
        if key == "updated_at" and isinstance(value, str):
            value = datetime.fromisoformat(value)
        if hasattr(node, key):
            object.__setattr__(node, key, value)
        else:
            node.properties[key] = value
    if "updated_at" not in updates:
        node.updated_at = datetime.now(UTC)
    return node

upsert_edge async

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

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

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

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight by 1 if (tenant, source, target, rel) matches."""
    edge.tenant_id = scope.tenant_id
    tenant_edges = self._edges[scope.tenant_id]
    for existing in tenant_edges.values():
        if (
            str(existing.source) == str(edge.source)
            and str(existing.target) == str(edge.target)
            and existing.relationship == edge.relationship
        ):
            object.__setattr__(existing, "weight", existing.weight + 1)
            return existing
    tenant_edges[str(edge.id)] = edge
    return edge

InMemoryVectorBackend

InMemoryVectorBackend()

In-memory implementation of the VectorBackend protocol.

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

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

def __init__(self) -> None:
    self._store: dict[str, list[dict[str, Any]]] = defaultdict(list)

add async

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

Add vectors with metadata to the store.

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

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

async def add(
    self,
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None:
    """Add vectors with metadata to the store.

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

count async

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

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

async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    return len(self._store.get(scope.tenant_id, []))

delete async

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

Delete vectors by their IDs.

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

async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    id_set = set(ids)
    if scope.tenant_id in self._store:
        self._store[scope.tenant_id] = [
            e for e in self._store[scope.tenant_id] if e["id"] not in id_set
        ]

search async

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

Brute-force cosine similarity search.

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

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

async def search(
    self,
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]:
    """Brute-force cosine similarity search.

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