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 src/symfonic/memory/protocols.py

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 async

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 src/symfonic/memory/protocols.py

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 (Jarvio 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 src/symfonic/memory/protocols.py

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 (Jarvio 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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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."""
    ...

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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).
    """
    ...

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.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."""
    ...

count async

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

Source code in src/symfonic/memory/protocols.py

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 src/symfonic/memory/protocols.py

async def delete(self, scope: TenantScope, ids: list[str]) -> None:
    """Delete vectors by their IDs."""
    ...

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 src/symfonic/memory/protocols.py

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.
    """
    ...