Skip to content

symfonic.memory.protocols

protocols

Protocol interfaces defining the stable public contract.

Architecture: BaseMemoryStore is the UNIFORM protocol. Every memory layer MUST implement all five methods (retrieve, write, summarize, link, delete). Methods that are not meaningful for a given layer return no-op results: - retrieve -> empty list - write -> no-op - summarize -> empty string - link -> no-op - delete -> no-op

Layers also expose layer-specific domain methods (store_fact, query_events, push, set_trigger, etc.) which are the rich API used by MemoryOrchestrator.commit_pending via layer-type dispatch.

Backend protocols (GraphBackend, VectorBackend, EmbeddingProvider) define the storage abstraction that concrete backends implement.

BaseMemoryStore

Bases: Protocol

Uniform protocol that every memory layer must implement.

This is the adapter interface. Layers that do not support a given operation return no-op results rather than raising NotImplementedError.

delete async 

delete(scope: TenantScope, node_id: NodeId) -> None

Delete a memory entry/node from this layer.

No-op if deletion is not meaningful for this layer.

Source code in symfonic/memory/protocols.py 
92
93
94
95
96
97
async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete a memory entry/node from this layer.

    No-op if deletion is not meaningful for this layer.
    """
    ...

link(scope: TenantScope, source: NodeId, target: NodeId, relationship: str) -> None

Create a relationship link between two nodes.

No-op if linking is not meaningful for this layer.

Source code in symfonic/memory/protocols.py 
79
80
81
82
83
84
85
86
87
88
89
90
async def link(
    self,
    scope: TenantScope,
    source: NodeId,
    target: NodeId,
    relationship: str,
) -> None:
    """Create a relationship link between two nodes.

    No-op if linking is not meaningful for this layer.
    """
    ...

render_for_prompt 

render_for_prompt(entry: Any, scrubber: Callable[[dict[str, Any]], dict[str, Any]] | None = None) -> str

Render a single retrieved entry for the MEMORY_CONTEXT block.

v7.7 (adopter procedural-render-bypass follow-up). Each layer owns its prompt-render contract so the engine's hydrate loop becomes polymorphic dispatch (one line, no schema-baked key enumeration). Layers that don't need a layer-specific shape delegate to :func:symfonic.memory.render.render_legacy -- that helper preserves the pre-v7.7 [layer] content (k=v) byte-identical output so prompt-cache parity holds for adopters who haven't opted into any new metadata surface.

scrubber is the agent-bound credential-scrubber callable (SymfonicAgent._scrub_props). Layers MUST thread it into any properties rendering so v6.1.9 F2 credential-scrubbing defense-in-depth holds through every render path. None when the agent has no credential pattern configured.

Returns the fully-rendered line including the leading [<layer>] tag. The engine assembles the final MEMORY_CONTEXT by joining all per-entry lines with \n.

Source code in symfonic/memory/protocols.py 
 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
def render_for_prompt(
    self,
    entry: Any,
    scrubber: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
) -> str:
    """Render a single retrieved entry for the ``MEMORY_CONTEXT`` block.

    v7.7 (adopter procedural-render-bypass follow-up).  Each layer
    owns its prompt-render contract so the engine's hydrate loop
    becomes polymorphic dispatch (one line, no schema-baked key
    enumeration).  Layers that don't need a layer-specific shape
    delegate to :func:`symfonic.memory.render.render_legacy` --
    that helper preserves the pre-v7.7 ``[layer] content (k=v)``
    byte-identical output so prompt-cache parity holds for
    adopters who haven't opted into any new metadata surface.

    ``scrubber`` is the agent-bound credential-scrubber callable
    (``SymfonicAgent._scrub_props``).  Layers MUST thread it into
    any ``properties`` rendering so v6.1.9 F2 credential-scrubbing
    defense-in-depth holds through every render path.  ``None``
    when the agent has no credential pattern configured.

    Returns the fully-rendered line including the leading
    ``[<layer>]`` tag.  The engine assembles the final
    ``MEMORY_CONTEXT`` by joining all per-entry lines with ``\\n``.
    """
    ...

retrieve async 

retrieve(scope: TenantScope, query: str, top_k: int = 5) -> list[Any]

Retrieve memory entries matching the query.

Returns an empty list if retrieval is not meaningful for this layer.

Source code in symfonic/memory/protocols.py 
53
54
55
56
57
58
59
60
61
62
63
async def retrieve(
    self,
    scope: TenantScope,
    query: str,
    top_k: int = 5,
) -> list[Any]:
    """Retrieve memory entries matching the query.

    Returns an empty list if retrieval is not meaningful for this layer.
    """
    ...

summarize async 

summarize(scope: TenantScope, entries: list[Any]) -> str

Produce a summary of the given entries.

Returns an empty string if summarization is not meaningful.

Source code in symfonic/memory/protocols.py 
72
73
74
75
76
77
async def summarize(self, scope: TenantScope, entries: list[Any]) -> str:
    """Produce a summary of the given entries.

    Returns an empty string if summarization is not meaningful.
    """
    ...

write async 

write(scope: TenantScope, entry: Any) -> None

Write a memory entry to this layer.

No-op if writing is not meaningful for this layer.

Source code in symfonic/memory/protocols.py 
65
66
67
68
69
70
async def write(self, scope: TenantScope, entry: Any) -> None:
    """Write a memory entry to this layer.

    No-op if writing is not meaningful for this layer.
    """
    ...

EmbeddingProvider

Bases: Protocol

Protocol for embedding text into vector representations.

embed async 

embed(text: str) -> list[float]

Embed a single text string into a vector.

Source code in symfonic/memory/protocols.py 
132
133
134
async def embed(self, text: str) -> list[float]:
    """Embed a single text string into a vector."""
    ...

embed_batch async 

embed_batch(texts: list[str]) -> list[list[float]]

Embed multiple texts into vectors.

Source code in symfonic/memory/protocols.py 
136
137
138
async def embed_batch(self, texts: list[str]) -> list[list[float]]:
    """Embed multiple texts into vectors."""
    ...

GraphBackend

Bases: Protocol

Protocol for graph node and edge persistence.

All methods enforce tenant isolation via TenantScope. Provides 9 core operations for full graph lifecycle management.

add_edge async 

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

Add an edge to the graph. Returns the persisted edge.

Source code in symfonic/memory/protocols_graph.py 
34
35
36
async def add_edge(self, scope: TenantScope, edge: Any) -> Any:
    """Add an edge to the graph. Returns the persisted edge."""
    ...

add_node async 

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

Add a node to the graph. Returns the persisted node.

Source code in symfonic/memory/protocols_graph.py 
30
31
32
async def add_node(self, scope: TenantScope, node: Any) -> Any:
    """Add a node to the graph. Returns the persisted node."""
    ...

delete_edge async 

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

Delete a single edge by its ID.

Source code in symfonic/memory/protocols_graph.py 
90
91
92
async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    ...

delete_node async 

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

Delete a node. When cascade=True, also removes all connected edges.

Source code in symfonic/memory/protocols_graph.py 
81
82
83
84
85
86
87
88
async def delete_node(
    self,
    scope: TenantScope,
    node_id: NodeId,
    cascade: bool = False,
) -> None:
    """Delete a node. When cascade=True, also removes all connected edges."""
    ...

delete_subtree async 

delete_subtree(scope: TenantScope) -> int

Erase every node at scope and below it. Returns the node count.

The descendant direction, and the only method here that runs it. Every read above is prefix-visibility-scoped: it answers "what can a query at P see?", which returns P's ancestors. Nothing enumerates what lies under a scope, so MemoryLifecyclePort.forget — which must erase a scope and its descendants — had no call to make.

One backend-native operation, never enumerate-then-delete. Two calls leave a window in which a concurrent write lands in a scope that was discovered but not yet swept, and that window is open during precisely the operation that must not miss a row: a tenant deletion while other work is still draining. Backends express it as a single statement (see :mod:symfonic.memory.subtree for the shared predicate); the in-memory backend does it inside one synchronous span.

Connected edges go with the nodes, as delete_node(cascade=True) does — an erased memory that left its edges behind would keep a traversable stub of what was supposed to be gone.

Source code in symfonic/memory/protocols_graph.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
async def delete_subtree(self, scope: TenantScope) -> int:
    """Erase every node at ``scope`` and below it. Returns the node count.

    The *descendant* direction, and the only method here that runs it.
    Every read above is prefix-visibility-scoped: it answers "what can a
    query at P see?", which returns P's **ancestors**. Nothing enumerates
    what lies under a scope, so ``MemoryLifecyclePort.forget`` — which must
    erase a scope and its descendants — had no call to make.

    **One backend-native operation, never enumerate-then-delete.** Two
    calls leave a window in which a concurrent write lands in a scope that
    was discovered but not yet swept, and that window is open during
    precisely the operation that must not miss a row: a tenant deletion
    while other work is still draining. Backends express it as a single
    statement (see :mod:`symfonic.memory.subtree` for the shared
    predicate); the in-memory backend does it inside one synchronous span.

    Connected edges go with the nodes, as ``delete_node(cascade=True)``
    does — an erased memory that left its edges behind would keep a
    traversable stub of what was supposed to be gone.
    """
    ...

get_neighbors async 

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

Get neighboring nodes, optionally filtered by relationship type.

Source code in symfonic/memory/protocols_graph.py 
42
43
44
45
46
47
48
49
async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
) -> list[Any]:
    """Get neighboring nodes, optionally filtered by relationship type."""
    ...

get_node async 

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

Get a node by ID. Returns None if not found.

Source code in symfonic/memory/protocols_graph.py 
38
39
40
async def get_node(self, scope: TenantScope, node_id: NodeId) -> Any | None:
    """Get a node by ID. Returns None if not found."""
    ...

query_edges async 

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

Query edges with optional filters and pagination.

Parameters:

Name Type Description Default
scope TenantScope

Tenant isolation scope.

 required
filters dict[str, Any] | None

Optional filter dict. Supported key: relationship (exact match).

 None
limit int

Maximum number of edges to return.

 50
offset int

Number of edges to skip (for pagination).

 0

Returns:

Type Description
list[Any]

List of MemoryEdge objects ordered by created_at descending.
Source code in symfonic/memory/protocols_graph.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
async def query_edges(
    self,
    scope: TenantScope,
    filters: dict[str, Any] | None = None,
    limit: int = 50,
    offset: int = 0,
) -> list[Any]:
    """Query edges with optional filters and pagination.

    Args:
        scope: Tenant isolation scope.
        filters: Optional filter dict. Supported key: ``relationship`` (exact match).
        limit: Maximum number of edges to return.
        offset: Number of edges to skip (for pagination).

    Returns:
        List of ``MemoryEdge`` objects ordered by ``created_at`` descending.
    """
    ...

query_nodes async 

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

Query nodes by filters (layer, label, properties, etc.).

limit=None returns all matching nodes (no truncation).

Source code in symfonic/memory/protocols_graph.py 
60
61
62
63
64
65
66
67
68
69
70
async def query_nodes(
    self,
    scope: TenantScope,
    filters: dict[str, Any],
    limit: int | None = 50,
) -> list[Any]:
    """Query nodes by filters (layer, label, properties, etc.).

    ``limit=None`` returns all matching nodes (no truncation).
    """
    ...

query_subtree async 

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

Query nodes at scope and below it. The descendant read.

The mirror of :meth:query_nodes, which is prefix-visibility-scoped and so returns a query's ancestors. Both directions are needed and neither substitutes for the other: retrieval asks "what may this scope see?", while the lifecycle verbs ask "what did this scope and its children produce?" — flush(scope) must publish a pending write made in a descendant session, and no ancestor-scoped read can find one.

Same filter vocabulary as :meth:query_nodes; limit=None returns every match.

Source code in symfonic/memory/protocols_graph.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
async def query_subtree(
    self,
    scope: TenantScope,
    filters: dict[str, Any],
    limit: int | None = None,
) -> list[Any]:
    """Query nodes at ``scope`` and **below** it. The descendant read.

    The mirror of :meth:`query_nodes`, which is prefix-visibility-scoped
    and so returns a query's *ancestors*. Both directions are needed and
    neither substitutes for the other: retrieval asks "what may this scope
    see?", while the lifecycle verbs ask "what did this scope and its
    children produce?" — ``flush(scope)`` must publish a pending write made
    in a descendant session, and no ancestor-scoped read can find one.

    Same filter vocabulary as :meth:`query_nodes`; ``limit=None`` returns
    every match.
    """
    ...

traverse async 

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

Traverse the graph from a start node up to max_depth.

Source code in symfonic/memory/protocols_graph.py 
51
52
53
54
55
56
57
58
async def traverse(
    self,
    scope: TenantScope,
    start: NodeId,
    max_depth: int,
) -> list[Any]:
    """Traverse the graph from a start node up to max_depth."""
    ...

update_node async 

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

Partial update of node properties. Returns the updated node.

Source code in symfonic/memory/protocols_graph.py 
72
73
74
75
76
77
78
79
async def update_node(
    self,
    scope: TenantScope,
    node_id: NodeId,
    updates: dict[str, Any],
) -> Any:
    """Partial update of node properties. Returns the updated node."""
    ...

upsert_edge async 

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

Insert edge or increment weight if it already exists.

An edge is considered a duplicate when (tenant_id, source, target, relationship) all match. On conflict the weight is incremented by 1; no duplicate row is created and the edge id is preserved from the first insert.

Source code in symfonic/memory/protocols_graph.py 
157
158
159
160
161
162
163
164
async def upsert_edge(self, scope: TenantScope, edge: Any) -> Any:
    """Insert edge or increment weight if it already exists.

    An edge is considered a duplicate when (tenant_id, source, target, relationship)
    all match. On conflict the weight is incremented by 1; no duplicate row is
    created and the edge id is preserved from the first insert.
    """
    ...

VectorBackend

Bases: Protocol

Protocol for vector storage and similarity search.

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.

Source code in symfonic/memory/protocols.py 
145
146
147
148
149
150
151
152
153
154
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."""
    ...

count async 

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

Source code in symfonic/memory/protocols.py 
176
177
178
async def count(self, scope: TenantScope) -> int:
    """Return the total number of stored vectors for the tenant."""
    ...

delete async 

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

Delete vectors by their IDs.

Source code in symfonic/memory/protocols.py 
168
169
170
async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    ...

delete_subtree async 

delete_subtree(scope: TenantScope) -> int

Delete every vector at scope and below it.

Source code in symfonic/memory/protocols.py 
172
173
174
async def delete_subtree(self, scope: TenantScope) -> int:
    """Delete every vector at ``scope`` and below it."""
    ...

search async 

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

Search for the top_k most similar vectors.

Returns list of dicts with keys: id, score, metadata, document.

Source code in symfonic/memory/protocols.py 
156
157
158
159
160
161
162
163
164
165
166
async def search(
    self,
    scope: TenantScope,
    query_embedding: list[float],
    top_k: int = 5,
) -> list[dict[str, Any]]:
    """Search for the top_k most similar vectors.

    Returns list of dicts with keys: id, score, metadata, document.
    """
    ...