symfonic.memory

memory

symfonic.memory: Production-grade, async-first, multi-layer memory library.

Provides protocol-driven graph-aware memory with tenant isolation, multi-signal retrieval scoring, and capability routing. Memory consolidation is handled by SleepConsolidator (see :mod:symfonic.core.learning.consolidation).

AssembledContext

AssembledContext(**data: Any)

Bases: BaseModel

Fully assembled context ready for LLM consumption.

.. deprecated:: AssembledContext is deprecated. Hydration is handled by SymfonicAgent._hydrate() which returns HydratedMemory.

Source code in src/symfonic/memory/models/context.py

def __init__(self, **data: Any) -> None:
    warnings.warn(
        "AssembledContext is deprecated. Hydration is handled by "
        "SymfonicAgent._hydrate().",
        DeprecationWarning,
        stacklevel=2,
    )
    super().__init__(**data)

is_within_budget

is_within_budget(budget: ContextBudget) -> bool

Check whether this context respects all budget constraints.

Returns True if all component counts are within the budget limits. max_memory_entries was removed in v5.6.0 (v6.0 API freeze prep); memory-entry count is no longer budget-enforced -- the live SymfonicAgent._hydrate path uses max_tokens instead.

Source code in src/symfonic/memory/models/context.py

def is_within_budget(self, budget: ContextBudget) -> bool:
    """Check whether this context respects all budget constraints.

    Returns True if all component counts are within the budget limits.
    ``max_memory_entries`` was removed in v5.6.0 (v6.0 API freeze prep);
    memory-entry count is no longer budget-enforced -- the live
    ``SymfonicAgent._hydrate`` path uses ``max_tokens`` instead.
    """
    if len(self.selected_tools) > budget.max_tools:
        return False
    return not len(self.conversation_history) > budget.max_history_messages

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

CandidateNode

Bases: BaseModel

Intermediate model normalizing vector and graph results before scoring.

The RetrievalEngine converts all candidate results (whether from vector search or graph traversal) into CandidateNode instances. Hits without a graph node get a synthetic MemoryNode with access_count=0 and graph_proximity=0.0.

CapabilityRouter

CapabilityRouter(
    procedural_layer: ProceduralLayer,
    catalog: ToolCatalog,
    llm: Any,
    max_tools: int = 5,
    *,
    observability_hook: ObservabilityHook | None = None,
)

Routes user queries to the appropriate tools via LLM selection.

Combines procedural memory (learned skills) with a static tool catalog to select the minimal set of tools needed for a given query.

Source code in src/symfonic/memory/layers/procedural/router.py

def __init__(
    self,
    procedural_layer: ProceduralLayer,
    catalog: ToolCatalog,
    llm: Any,
    max_tools: int = 5,
    *,
    observability_hook: ObservabilityHook | None = None,
) -> None:
    self._procedural = procedural_layer
    self._catalog = catalog
    self._llm = llm
    self._max_tools = max_tools
    # v8.3.2 observability-bug fix (G5): the procedural-router LLM call
    # emitted ONLY on the merge-dependent callback manager (same bug
    # class as the v8.3.1 critic).  When a hook is supplied it also
    # emits on the always-on ``ObservabilityHook``.  ``None`` resolves
    # to ``NoOpObservabilityHook`` so the default is byte-identical.
    # NOTE: the live caller (``MemoryOrchestrator``) does NOT yet thread
    # a real hook -- see v8.3.2 report; this lets direct callers wire
    # one without invasive orchestrator plumbing.
    if observability_hook is None:
        from symfonic.core.observability.hooks import NoOpObservabilityHook

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

route async

route(
    scope: TenantScope,
    query: str,
    conversation_history: list[Any] | None = None,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "procedural_router",
) -> RoutingResult

Route a query to appropriate tools.

Steps: 1. Query procedural layer for relevant skills. 2. Search catalog for matching tools. 3. Use LLM with structured output to select top tools. 4. Return only minimal schemas for selected tools.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
query	str	The user's query or request.	required
conversation_history	list[Any] | None	Optional recent conversation for context.	None
callback_manager	CallbackManager | None	v7.4.3 (Jarvio Ask 5) -- when provided, fires on_llm_end(node_name="procedural_router") after the LLM call so the (opt-in) capability router's tokens are visible to billing / OTel. None preserves byte- identical pre-7.4.3 behaviour.	None
run_id	str	Engine run identifier for callback correlation.	''
model_name	str	Optional model identifier stamped on the LLMEndEvent. Defaults to "procedural_router".	'procedural_router'

Returns:

Type	Description
RoutingResult	RoutingResult with intent, selected tools, and reasoning.

Source code in src/symfonic/memory/layers/procedural/router.py

async def route(
    self,
    scope: TenantScope,
    query: str,
    conversation_history: list[Any] | None = None,
    *,
    callback_manager: CallbackManager | None = None,
    run_id: str = "",
    model_name: str = "procedural_router",
) -> RoutingResult:
    """Route a query to appropriate tools.

    Steps:
    1. Query procedural layer for relevant skills.
    2. Search catalog for matching tools.
    3. Use LLM with structured output to select top tools.
    4. Return only minimal schemas for selected tools.

    Args:
        scope: Tenant scope for isolation.
        query: The user's query or request.
        conversation_history: Optional recent conversation for context.
        callback_manager: v7.4.3 (Jarvio Ask 5) -- when provided, fires
            ``on_llm_end(node_name="procedural_router")`` after the
            LLM call so the (opt-in) capability router's tokens are
            visible to billing / OTel.  ``None`` preserves byte-
            identical pre-7.4.3 behaviour.
        run_id: Engine run identifier for callback correlation.
        model_name: Optional model identifier stamped on the
            LLMEndEvent.  Defaults to ``"procedural_router"``.

    Returns:
        RoutingResult with intent, selected tools, and reasoning.
    """
    # Step 1: Query procedural layer for relevant skills
    skills = await self._procedural.query_skills(scope, query, top_k=5)
    skill_context = _format_skill_context(skills)

    # Step 2: Format tool catalog
    all_tools = self._catalog.list_all()
    catalog_text = "\n".join(
        f"- {t['tool_id']}: {t['description']}" for t in all_tools
    ) or "No tools registered."

    # Step 3: Build prompt and call LLM
    prompt_template = load_prompt("router")
    prompt = prompt_template.format(
        max_tools=self._max_tools,
        tool_catalog=catalog_text,
        skill_context=skill_context,
    )

    try:
        from symfonic.core.callbacks.emit import (
            _extract_usage,
            emit_llm_end_from_result,
            emit_llm_start,
            llm_timing,
            resolve_model_name,
        )

        _resolved_model = resolve_model_name(self._llm, model_name)
        _router_prompt = prompt + f"\n\nUser query: {query}"

        # v8.3.2 (G5): emit on the always-on ObservabilityHook BEFORE
        # the call (mirrors react.py:1078 / metacognitive.py:412).
        try:
            await self._observability_hook.on_llm_start(
                _resolved_model, [_router_prompt], run_id
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_start failed")

        # v8.4.0 CALLBACK-PATH fix: open the OTel span on the merged
        # manager BEFORE the ainvoke so the existing END (:306) closes
        # it.  Same node_name; byte-identical no-op when no manager.
        if callback_manager is not None and not callback_manager.is_noop:
            await emit_llm_start(
                callback_manager,
                model=_resolved_model,
                messages=[_router_prompt],
                run_id=run_id,
                node_name="procedural_router",
            )

        # v7.15.0: capture timing so LLMEndEvent carries duration.
        async with llm_timing() as _timing:
            response = await self._llm.ainvoke(_router_prompt)

        # v8.3.2 (G5): emit on the always-on ObservabilityHook AFTER
        # the call (mirrors react.py:1433 / metacognitive.py:447).
        try:
            await self._observability_hook.on_llm_end(
                _resolved_model,
                str(getattr(response, "content", response)),
                _extract_usage(response),
                run_id,
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_end failed")

        # v7.4.3 Jarvio Ask 5: emit on_llm_end so the (opt-in)
        # LLM-backed router's tokens show up in billing / OTel.
        # v7.14.0: stamp the resolved SKU instead of the
        # caller-passed name.  See callbacks/emit.py:resolve_model_name
        # for the provider-attribute survey + fallback semantics.
        await emit_llm_end_from_result(
            callback_manager,
            model=_resolved_model,
            result=response,
            run_id=run_id,
            node_name="procedural_router",
            timing=_timing,
        )

        content = response.content if hasattr(response, "content") else str(response)
        return self._parse_response(content, all_tools)
    except Exception:
        logger.exception("Router LLM call failed for tenant %s", scope.tenant_id)
        return RoutingResult(intent=query, reasoning="LLM routing failed, no tools selected")

ContextBudget

ContextBudget(**data: Any)

Bases: BaseModel

Defines the limits for context assembly.

.. deprecated:: ContextBudget is deprecated. Budget constraints are now handled internally by SymfonicAgent._hydrate() via FrameworkConfig.

Source code in src/symfonic/memory/models/context.py

def __init__(self, **data: Any) -> None:
    warnings.warn(
        "ContextBudget is deprecated. Budget constraints are handled internally "
        "by SymfonicAgent._hydrate() via FrameworkConfig.",
        DeprecationWarning,
        stacklevel=2,
    )
    super().__init__(**data)

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

GraphMemoryStore

GraphMemoryStore(
    backend: GraphBackend,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
)

Graph-aware memory store wrapping a GraphBackend.

All operations enforce tenant isolation via TenantScope. Access counts are automatically incremented on reads to support frequency-based retrieval scoring.

v7.3 Item 13.1: optional auto-embed at the store level.

The graph-backed memory LAYERS (semantic / procedural / prospective) wrap their own writes through :func:~symfonic.memory.embeddings.auto_embed.maybe_embed so layer-API callers benefit even when constructing layers without the orchestrator. Consolidation phases that bypass the layers (Item 12 EntityLinker calls GraphMemoryStore.add_node directly) need the same auto-embed pathway -- that's what the constructor kwargs here are for.

The double wrap (layer -> store) is idempotent: maybe_embed short-circuits when node.embedding is already populated, so a node that went through SemanticLayer.store_fact only triggers one provider call regardless of how many layers of wrapping it traverses.

Both kwargs default to None so legacy callers that build GraphMemoryStore directly are byte-identical to v7.2.

Source code in src/symfonic/memory/graph/store.py

def __init__(
    self,
    backend: GraphBackend,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
) -> None:
    """v7.3 Item 13.1: optional auto-embed at the store level.

    The graph-backed memory LAYERS (semantic / procedural /
    prospective) wrap their own writes through
    :func:`~symfonic.memory.embeddings.auto_embed.maybe_embed` so
    layer-API callers benefit even when constructing layers without
    the orchestrator. Consolidation phases that bypass the layers
    (Item 12 EntityLinker calls ``GraphMemoryStore.add_node``
    directly) need the same auto-embed pathway -- that's what the
    constructor kwargs here are for.

    The double wrap (layer -> store) is idempotent: ``maybe_embed``
    short-circuits when ``node.embedding`` is already populated, so
    a node that went through ``SemanticLayer.store_fact`` only
    triggers one provider call regardless of how many layers of
    wrapping it traverses.

    Both kwargs default to ``None`` so legacy callers that build
    ``GraphMemoryStore`` directly are byte-identical to v7.2.
    """
    self._backend = backend
    self._embedding_provider = embedding_provider
    self._embedding_cache = embedding_cache

add_edge async

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

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

Source code in src/symfonic/memory/graph/store.py

async def add_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Add an edge to the graph. Returns the persisted edge."""
    edge.tenant_id = scope.tenant_id
    return await self._backend.add_edge(scope, edge)

add_node async

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

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

v7.3 Item 13.1: when an embedding_provider is wired into this store, nodes the caller left with embedding=None auto-embed before the backend write. Caller-provided embeddings always win (see :func:maybe_embed).

Source code in src/symfonic/memory/graph/store.py

async def add_node(self, scope: TenantScope, node: MemoryNode) -> MemoryNode:
    """Add a node to the graph. Returns the persisted node.

    v7.3 Item 13.1: when an ``embedding_provider`` is wired into
    this store, nodes the caller left with ``embedding=None``
    auto-embed before the backend write. Caller-provided embeddings
    always win (see :func:`maybe_embed`).
    """
    node.tenant_id = scope.tenant_id
    if self._embedding_provider is not None:
        # Late import keeps the symfonic.memory.embeddings package an
        # optional surface for callers that never construct one.
        from symfonic.memory.embeddings.auto_embed import maybe_embed

        node = await maybe_embed(
            node, self._embedding_provider, self._embedding_cache,
        )
    return await self._backend.add_node(scope, node)

bump_spreading async

bump_spreading(
    scope: TenantScope,
    node_id: NodeId,
    *,
    include_retracted: bool = False,
) -> MemoryNode | None

Fetch a node and increment its spreading_access_count.

v6.2 T02: BFS-induced reads funnel through this helper instead of :meth:get_node, so the Phase 1 recurrence signal can distinguish direct fetches from one-shot spreading-activation visits. Both counters carry independent monotonic semantics: - access_count grows on get_node - spreading_access_count grows on bump_spreading

The method is intentionally named bump_spreading (rather than get_node_via_spreading or a private _bump_spreading) so alternative traversal strategies in future sprints can call it directly without reaching into private API.

Returns None if the node does not exist or belongs to another tenant; otherwise returns the updated node.

Soft-retracted nodes return None by default (and are not bumped) so a retracted false memory cannot re-enter context via spreading activation -- notably as a live node's neighbor. Pass include_retracted=True only for audit/maintenance reads.

Source code in src/symfonic/memory/graph/store.py

async def bump_spreading(
    self, scope: TenantScope, node_id: NodeId,
    *, include_retracted: bool = False,
) -> MemoryNode | None:
    """Fetch a node and increment its spreading_access_count.

    v6.2 T02: BFS-induced reads funnel through this helper instead
    of :meth:`get_node`, so the Phase 1 recurrence signal can
    distinguish direct fetches from one-shot spreading-activation
    visits. Both counters carry independent monotonic semantics:
    - access_count grows on ``get_node``
    - spreading_access_count grows on ``bump_spreading``

    The method is intentionally named ``bump_spreading`` (rather than
    ``get_node_via_spreading`` or a private ``_bump_spreading``) so
    alternative traversal strategies in future sprints can call it
    directly without reaching into private API.

    Returns None if the node does not exist or belongs to another
    tenant; otherwise returns the updated node.

    Soft-retracted nodes return ``None`` by default (and are not bumped) so
    a retracted false memory cannot re-enter context via spreading
    activation -- notably as a live node's neighbor. Pass
    ``include_retracted=True`` only for audit/maintenance reads.
    """
    node = await self._backend.get_node(scope, node_id)
    if node is None:
        return None
    if not include_retracted and is_retracted(node.properties):
        return None
    updated = await self._backend.update_node(
        scope,
        node_id,
        {
            "spreading_access_count": node.spreading_access_count + 1,
            "updated_at": datetime.now(UTC).isoformat(),
        },
    )
    return updated

delete_edge async

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

Delete a single edge by its ID.

Source code in src/symfonic/memory/graph/store.py

async def delete_edge(self, scope: TenantScope, edge_id: EdgeId) -> None:
    """Delete a single edge by its ID."""
    await self._backend.delete_edge(scope, edge_id)

delete_node async

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

Delete a node and cascade-delete all connected edges.

Source code in src/symfonic/memory/graph/store.py

async def delete_node(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete a node and cascade-delete all connected edges."""
    await self._backend.delete_node(scope, node_id, cascade=True)

distinct_labels async

distinct_labels(
    scope: TenantScope, *, include_retracted: bool = False
) -> list[dict[str, Any]]

Get distinct labels with counts and last_updated.

Aggregates all nodes in the tenant scope by their label field.

Soft-retracted nodes are excluded by default so a label whose only node has been retracted is NOT reported as present -- otherwise a required-memory discovery check (retrieval.discovery) would mark a block satisfied when no live node backs it. Pass include_retracted=True for audit aggregation.

Returns:

Type	Description
list[dict[str, Any]]	List of dicts with keys label, count, last_updated.

Source code in src/symfonic/memory/graph/store.py

async def distinct_labels(
    self, scope: TenantScope, *, include_retracted: bool = False,
) -> list[dict[str, Any]]:
    """Get distinct labels with counts and last_updated.

    Aggregates all nodes in the tenant scope by their ``label`` field.

    Soft-retracted nodes are excluded by default so a label whose only
    node has been retracted is NOT reported as present -- otherwise a
    required-memory discovery check (``retrieval.discovery``) would mark a
    block satisfied when no live node backs it. Pass
    ``include_retracted=True`` for audit aggregation.

    Returns:
        List of dicts with keys ``label``, ``count``, ``last_updated``.
    """
    # Full aggregation, no cap: route through the store's own
    # ``query_nodes`` (the single retraction choke point). A flat
    # backend cap was unsafe here -- with retractions excluded AFTER a
    # truncated fetch, N>=cap retracted rows ahead of a live node made
    # its label vanish (retrieval.discovery would then mark a satisfied
    # block UNsatisfied). ``limit=None`` scans every node and refills
    # past retracted rows via the same doubling over-fetch as siblings.
    nodes = await self.query_nodes(
        scope, limit=None, include_retracted=include_retracted,
    )
    groups: dict[str, dict[str, Any]] = defaultdict(
        lambda: {"count": 0, "last_updated": None},
    )
    for node in nodes:
        g = groups[node.label]
        g["count"] += 1
        if node.updated_at and (
            g["last_updated"] is None or node.updated_at > g["last_updated"]
        ):
            g["last_updated"] = node.updated_at
    return [
        {"label": label, "count": g["count"], "last_updated": g["last_updated"]}
        for label, g in groups.items()
    ]

get_neighbors async

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

Get neighboring nodes, optionally filtered by relationship type.

Soft-retracted neighbours are excluded by default so a retracted node cannot re-enter context as a graph neighbour of a live node. Audit / maintenance callers pass include_retracted=True.

Source code in src/symfonic/memory/graph/store.py

async def get_neighbors(
    self,
    scope: TenantScope,
    node_id: NodeId,
    relationship: str | None = None,
    *,
    include_retracted: bool = False,
) -> list[MemoryNode]:
    """Get neighboring nodes, optionally filtered by relationship type.

    Soft-retracted neighbours are excluded by default so a retracted node
    cannot re-enter context as a graph neighbour of a live node. Audit /
    maintenance callers pass ``include_retracted=True``.
    """
    neighbors = await self._backend.get_neighbors(
        scope, node_id, relationship,
    )
    if include_retracted:
        return neighbors
    return [n for n in neighbors if not is_retracted(n.properties)]

get_node async

get_node(
    scope: TenantScope,
    node_id: NodeId,
    *,
    include_retracted: bool = False,
) -> MemoryNode | None

Get a node by ID, incrementing its access count.

Returns None if the node does not exist or belongs to another tenant.

Soft-retracted nodes (see symfonic.memory.retraction for the namespaced marker contract) return None by default so a corrected/false-positive memory never re-enters context through a direct fetch; the access count is NOT bumped for them. Audit / erasure / prune callers pass include_retracted=True to reach them.

Source code in src/symfonic/memory/graph/store.py

async def get_node(
    self, scope: TenantScope, node_id: NodeId,
    *, include_retracted: bool = False,
) -> MemoryNode | None:
    """Get a node by ID, incrementing its access count.

    Returns None if the node does not exist or belongs to another tenant.

    Soft-retracted nodes (see ``symfonic.memory.retraction`` for the
    namespaced marker contract) return ``None`` by default so a
    corrected/false-positive memory never re-enters context through a
    direct fetch; the access count is NOT bumped for them. Audit /
    erasure / prune callers pass ``include_retracted=True`` to reach them.
    """
    node = await self._backend.get_node(scope, node_id)
    if node is None:
        return None
    if not include_retracted and is_retracted(node.properties):
        return None
    # Increment access_count for frequency scoring
    updated = await self._backend.update_node(
        scope,
        node_id,
        {
            "access_count": node.access_count + 1,
            "updated_at": datetime.now(UTC).isoformat(),
        },
    )
    return updated

list_edges async

list_edges(
    scope: TenantScope,
    *,
    limit: int = 50,
    offset: int = 0,
    relationship: str | None = None,
) -> list[MemoryEdge]

List edges for the tenant with optional filtering and pagination.

Delegates directly to GraphBackend.query_edges — no node iteration.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant isolation scope.	required
limit	int	Maximum number of edges to return (default 50).	50
offset	int	Number of edges to skip for pagination (default 0).	0
relationship	str | None	If provided, only edges of this type are returned.	None

Source code in src/symfonic/memory/graph/store.py

async def list_edges(
    self,
    scope: TenantScope,
    *,
    limit: int = 50,
    offset: int = 0,
    relationship: str | None = None,
) -> list[MemoryEdge]:
    """List edges for the tenant with optional filtering and pagination.

    Delegates directly to ``GraphBackend.query_edges`` — no node iteration.

    Args:
        scope: Tenant isolation scope.
        limit: Maximum number of edges to return (default 50).
        offset: Number of edges to skip for pagination (default 0).
        relationship: If provided, only edges of this type are returned.
    """
    filters: dict[str, Any] | None = None
    if relationship is not None:
        filters = {"relationship": relationship}
    return await self._backend.query_edges(scope, filters, limit, offset)

query_nodes async

query_nodes(
    scope: TenantScope,
    layer: MemoryLayer | None = None,
    label: str | None = None,
    label_prefix: str | None = None,
    filters: dict[str, object] | None = None,
    limit: int | None = None,
    include_retracted: bool = False,
) -> list[MemoryNode]

Query nodes by layer, label, label_prefix, and/or arbitrary filters.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant isolation scope.	required
layer	MemoryLayer | None	Exact match on the memory layer.	None
label	str | None	Exact match on the node label.	None
label_prefix	str | None	Prefix match on the node label (e.g. "SOUL" matches "SOUL: Joy" and "SOUL: name: Amiel"). label and label_prefix can coexist (AND logic).	None
filters	dict[str, object] | None	Arbitrary key/value filters passed through to the backend.	None
limit	int | None	Maximum rows to return. None (default) returns all matching nodes. Prior to v8.7.2 this wrapper silently dropped the caller's request and the backend truncated every read to its ~50-row default — starving retrieval scans, dedupe, and (worst) inverting working_graph_retention >= 50 pruning.	None
include_retracted	bool	When False (default) nodes soft-retracted via a retract_node extraction op (see symfonic.memory.retraction for the namespaced marker) are filtered out. This is the single read-side choke point that keeps a corrected/false-positive memory out of hydration, routing, and dedup. Consolidation's prune phase passes True so it can find and delete them. See docs/guides/09-consolidation-and-deep-sleep.md.	False

A positive limit combined with include_retracted=False no longer under-returns when a retracted row happens to fall inside the backend's page: the wrapper over-fetches (doubling the requested window) and refills with live nodes until limit is satisfied or the backend is exhausted, then trims to exactly limit. This keeps limit semantics honest for every backend (InMemory/Postgres/ Mongo) without any backend-side change.

Source code in src/symfonic/memory/graph/store.py

async def query_nodes(
    self,
    scope: TenantScope,
    layer: MemoryLayer | None = None,
    label: str | None = None,
    label_prefix: str | None = None,
    filters: dict[str, object] | None = None,
    limit: int | None = None,
    include_retracted: bool = False,
) -> list[MemoryNode]:
    """Query nodes by layer, label, label_prefix, and/or arbitrary filters.

    Args:
        scope: Tenant isolation scope.
        layer: Exact match on the memory layer.
        label: Exact match on the node label.
        label_prefix: Prefix match on the node label (e.g. ``"SOUL"``
            matches ``"SOUL: Joy"`` and ``"SOUL: name: Amiel"``).
            ``label`` and ``label_prefix`` can coexist (AND logic).
        filters: Arbitrary key/value filters passed through to the backend.
        limit: Maximum rows to return.  ``None`` (default) returns *all*
            matching nodes.  Prior to v8.7.2 this wrapper silently dropped
            the caller's request and the backend truncated every read to
            its ~50-row default — starving retrieval scans, dedupe, and
            (worst) inverting ``working_graph_retention >= 50`` pruning.
        include_retracted: When ``False`` (default) nodes soft-retracted
            via a ``retract_node`` extraction op (see
            ``symfonic.memory.retraction`` for the namespaced marker) are
            filtered out. This is the single read-side choke point that
            keeps a corrected/false-positive memory out of hydration,
            routing, and dedup. Consolidation's prune phase passes
            ``True`` so it can find and delete them. See
            ``docs/guides/09-consolidation-and-deep-sleep.md``.

    A positive ``limit`` combined with ``include_retracted=False`` no
    longer under-returns when a retracted row happens to fall inside the
    backend's page: the wrapper over-fetches (doubling the requested
    window) and refills with live nodes until ``limit`` is satisfied or
    the backend is exhausted, then trims to exactly ``limit``. This keeps
    ``limit`` semantics honest for every backend (InMemory/Postgres/
    Mongo) without any backend-side change.
    """
    filter_dict: dict[str, object] = {}
    if layer is not None:
        filter_dict["layer"] = layer.value
    if label is not None:
        filter_dict["label"] = label
    if label_prefix is not None:
        filter_dict["label_prefix"] = label_prefix
    if filters:
        filter_dict.update(filters)
    # v8.7.1: unify limit==0 semantics across backends. ``None`` = all,
    # a positive int caps the result, 0 means "no rows".  Mongo's native
    # ``.limit(0)`` otherwise means *unlimited* (the opposite) and InMemory
    # returned 1 row — so short-circuit here before dispatch.
    if limit == 0:
        return []
    if include_retracted:
        return await self._backend.query_nodes(scope, filter_dict, limit=limit)
    if limit is None:
        nodes = await self._backend.query_nodes(scope, filter_dict, limit=None)
        return [n for n in nodes if not is_retracted(n.properties)]
    # Retraction exclusion: a soft-retracted node stays in the store (for
    # audit + reversibility) but must never surface to any reader. Applied
    # here so every consumer of query_nodes -- semantic/procedural
    # retrieve, dedup, spreading activation -- inherits the exclusion for
    # free. Over-fetch with a doubling window so a retracted row inside
    # the backend's page doesn't silently starve the caller's ``limit``.
    fetch = limit
    while True:
        batch = await self._backend.query_nodes(scope, filter_dict, limit=fetch)
        live = [n for n in batch if not is_retracted(n.properties)]
        if len(live) >= limit or len(batch) < fetch:
            break
        fetch *= 2
    return live[:limit]

update_node async

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

Partial update of node properties.

Source code in src/symfonic/memory/graph/store.py

async def update_node(
    self, scope: TenantScope, node_id: NodeId, properties: dict[str, object]
) -> MemoryNode:
    """Partial update of node properties."""
    return await self._backend.update_node(scope, node_id, properties)

upsert_edge async

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

Insert edge or increment weight if it already exists.

Uniqueness key: (tenant_id, source, target, relationship). Each repeated call on the same pair increments weight by 1.

Source code in src/symfonic/memory/graph/store.py

async def upsert_edge(self, scope: TenantScope, edge: MemoryEdge) -> MemoryEdge:
    """Insert edge or increment weight if it already exists.

    Uniqueness key: (tenant_id, source, target, relationship).
    Each repeated call on the same pair increments weight by 1.
    """
    edge.tenant_id = scope.tenant_id
    return await self._backend.upsert_edge(scope, edge)

upsert_edge_props async

upsert_edge_props(
    scope: TenantScope,
    edge_id: EdgeId,
    props: dict[str, Any],
) -> MemoryEdge | None

Update edge properties in-place.

Attempts backend.update_edge when available; otherwise falls back to delete+re-add. Returns the updated edge, or None on failure.

Source code in src/symfonic/memory/graph/store.py

async def upsert_edge_props(
    self,
    scope: TenantScope,
    edge_id: EdgeId,
    props: dict[str, Any],
) -> MemoryEdge | None:
    """Update edge properties in-place.

    Attempts backend.update_edge when available; otherwise falls back
    to delete+re-add. Returns the updated edge, or None on failure.
    """
    if hasattr(self._backend, "update_edge"):
        return await self._backend.update_edge(scope, edge_id, props)
    # Fallback: find edge, apply props, delete, re-add
    edges = await self._backend.query_edges(scope, None, 10_000, 0)
    target_edge = None
    for e in edges:
        if str(e.id) == str(edge_id):
            target_edge = e
            break
    if target_edge is None:
        return None
    valid_props = {k: v for k, v in props.items() if hasattr(target_edge, k)}
    updated_edge = target_edge.model_copy(update=valid_props)
    # Delete-then-add (v8.7.2): the previous add-then-delete order re-added
    # the edge under the SAME id and then deleted that id, permanently
    # destroying the edge — on InMemory the delete popped the just-written
    # key, and on Postgres ``add_edge`` is ``ON CONFLICT DO NOTHING`` so the
    # write was a no-op before the delete removed the original.  Deleting
    # first, then re-adding the mutated copy, is correct on every backend.
    await self._backend.delete_edge(scope, edge_id)
    await self._backend.add_edge(scope, updated_edge)
    return updated_edge

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]

MemoryEdge

Bases: BaseModel

A directed edge in the tenant-scoped memory graph.

Edges represent typed relationships between MemoryNodes, such as 'KNOWS', 'USED_IN', 'TRIGGERED_BY', etc. Weight can be used to express relationship strength.

MemoryEntry

Bases: BaseModel

A memory entry returned from retrieval or submitted for writing.

Represents a single piece of information from any memory layer, optionally linked to a graph node and carrying a retrieval score.

MemoryLayer

Bases: str, Enum

The five layers of the Pentad memory model.

Each layer serves a distinct cognitive function and is backed by an independent store that implements BaseMemoryStore.

EPISODIC class-attribute instance-attribute

EPISODIC = 'episodic'

Narrative events, scenarios, and timestamps (When/Where).

PROCEDURAL class-attribute instance-attribute

PROCEDURAL = 'procedural'

Skills, code snippets, and workflows (How).

PROSPECTIVE class-attribute instance-attribute

PROSPECTIVE = 'prospective'

Commitments, reminders, and pending tasks (Future).

SEMANTIC class-attribute instance-attribute

SEMANTIC = 'semantic'

Permanent facts and graph entities (What).

WORKING class-attribute instance-attribute

WORKING = 'working'

Active conversation context, session-scoped (Now).

MemoryNode

Bases: BaseModel

A node in the tenant-scoped memory graph.

Nodes represent facts, events, skills, triggers, or working context depending on their layer assignment. Each node carries an optional embedding vector for similarity search and tracks access frequency for retrieval scoring.

durability property

durability: Durability

Return the node's durability marker (v7.26.2 -- Shape C1).

Reads from properties['durability']. Pre-v7.26.2 nodes (and adopters who never set the key) get "durable" -- the safe default that preserves byte-identical promotion behaviour. An unknown persisted value falls back to "durable" so the consolidation gate never silently drops a row.

This is a computed property, NOT a model Field: it does not change the serialised shape (Contract A).

is_promotable

is_promotable() -> bool

Return True when the node is eligible for consolidation promotion.

Only "durable" nodes promote. "transient" and "expired" are gated out of Phases 10/11/12.

Source code in src/symfonic/memory/models/node.py

def is_promotable(self) -> bool:
    """Return True when the node is eligible for consolidation promotion.

    Only ``"durable"`` nodes promote.  ``"transient"`` and ``"expired"``
    are gated out of Phases 10/11/12.
    """
    return self.durability == "durable"

is_transient

is_transient() -> bool

Return True when the node is marked transient.

Source code in src/symfonic/memory/models/node.py

def is_transient(self) -> bool:
    """Return True when the node is marked transient."""
    return self.durability == "transient"

MemoryOperation

Bases: BaseModel

A pending memory mutation to be applied by the orchestrator.

Each operation targets a specific layer and action. The node and edge fields are populated depending on the action type.

MemoryOrchestrator

MemoryOrchestrator(
    config: OrchestratorConfig,
    retrieval_engine: RetrievalEngine,
    router: CapabilityRouter,
    layers: dict[MemoryLayer, BaseMemoryStore],
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
    graph_store: GraphMemoryStore | None = None,
)

Main entry point for the graph memory library.

Coordinates retrieval, routing, and layer management. Construct via from_config for automatic wiring of all internal components from backends and configuration.

Memory consolidation is handled by :class:symfonic.core.learning.consolidation.SleepConsolidator.

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

def __init__(
    self,
    config: OrchestratorConfig,
    retrieval_engine: RetrievalEngine,
    router: CapabilityRouter,
    layers: dict[MemoryLayer, BaseMemoryStore],
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
    graph_store: GraphMemoryStore | None = None,
) -> None:
    self._config = config
    self._retrieval = retrieval_engine
    self._router = router
    self._layers = layers
    self._embedder = embedding_provider
    # v7.26.2: the shared graph store backing the layers, used by
    # upgrade_durability for a backend-agnostic read-modify-write.
    self._graph = graph_store
    # v7.3 Item 13.1: the same cache instance is handed to every
    # graph-backed layer at construction time so that identical text
    # written through SemanticLayer + ProceduralLayer + ProspectiveLayer
    # only triggers one provider.embed() call. The cache is optional
    # (default None) so legacy callers that build their own
    # MemoryOrchestrator without the factory are unaffected.
    self._embedding_cache = embedding_cache
    self._extraction = ExtractionService()

commit_pending async

commit_pending(
    scope: TenantScope, operations: list[MemoryOperation]
) -> None

Apply pending operations to the appropriate layers.

Filters operations below the write_policy.min_importance_threshold before committing.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
operations	list[MemoryOperation]	List of operations to apply.	required

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

async def commit_pending(
    self,
    scope: TenantScope,
    operations: list[MemoryOperation],
) -> None:
    """Apply pending operations to the appropriate layers.

    Filters operations below the write_policy.min_importance_threshold
    before committing.

    Args:
        scope: Tenant scope for isolation.
        operations: List of operations to apply.
    """
    threshold = self._config.write_policy.min_importance_threshold

    for op in operations:
        if op.importance < threshold:
            logger.debug(
                "Skipping operation %s (importance %.1f < threshold %.1f)",
                op.action, op.importance, threshold,
            )
            continue

        if op.action == "noop":
            continue

        layer = self._layers.get(op.layer)
        if layer is None:
            logger.warning("Layer %s not enabled, skipping operation", op.layer)
            continue

        entry = MemoryEntry(
            layer=op.layer,
            tenant_id=scope.tenant_id,
            content=op.node.label if op.node else "",
            node_id=op.node.id if op.node else None,
            importance=op.importance,
            metadata=op.node.properties if op.node else {},
        )

        await layer.write(scope, entry)

extract_memories async

extract_memories(
    scope: TenantScope,
    interaction: dict[str, Any],
    llm: Any,
    *,
    callback_manager: Any = None,
    run_id: str = "",
) -> list[MemoryOperation]

Extract structured memory operations from a conversation turn.

Uses the LLM to analyze the interaction and produce MemoryOperations for each memory layer as appropriate.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
interaction	dict[str, Any]	Dict with 'user_message' and 'assistant_response'.	required
llm	Any	LLM instance for extraction.	required
callback_manager	Any	v7.4.3 (Jarvio Ask 5) -- forwarded to the underlying ExtractionService so the consolidation LLM call emits on_llm_end(node_name="consolidation_extractor"). None preserves byte-identical pre-7.4.3 behaviour.	None
run_id	str	Engine run identifier for callback correlation.	''

Returns:

Type	Description
list[MemoryOperation]	List of MemoryOperations to be applied via commit_pending.

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

async def extract_memories(
    self,
    scope: TenantScope,
    interaction: dict[str, Any],
    llm: Any,
    *,
    callback_manager: Any = None,
    run_id: str = "",
) -> list[MemoryOperation]:
    """Extract structured memory operations from a conversation turn.

    Uses the LLM to analyze the interaction and produce MemoryOperations
    for each memory layer as appropriate.

    Args:
        scope: Tenant scope for isolation.
        interaction: Dict with 'user_message' and 'assistant_response'.
        llm: LLM instance for extraction.
        callback_manager: v7.4.3 (Jarvio Ask 5) -- forwarded to the
            underlying ``ExtractionService`` so the consolidation
            LLM call emits ``on_llm_end(node_name="consolidation_extractor")``.
            ``None`` preserves byte-identical pre-7.4.3 behaviour.
        run_id: Engine run identifier for callback correlation.

    Returns:
        List of MemoryOperations to be applied via commit_pending.
    """
    return await self._extraction.extract(
        scope, interaction, llm,
        callback_manager=callback_manager,
        run_id=run_id,
    )

from_config classmethod

from_config(
    config: OrchestratorConfig,
    graph_backend: GraphBackend,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    llm: Any,
    catalog: ToolCatalog | None = None,
) -> MemoryOrchestrator

Wire all internal components from configuration and backends.

This is the primary DI entry point for library consumers.

Parameters:

Name	Type	Description	Default
config	OrchestratorConfig	Orchestrator configuration.	required
graph_backend	GraphBackend	Graph storage backend.	required
vector_backend	VectorBackend	Vector storage backend.	required
embedding_provider	EmbeddingProvider	Text embedding provider.	required
llm	Any	LLM instance (BaseChatModel from langchain-core).	required
catalog	ToolCatalog | None	Optional tool catalog. Creates empty one if not provided.	None

Returns:

Type	Description
MemoryOrchestrator	Fully wired MemoryOrchestrator instance.

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

@classmethod
def from_config(
    cls,
    config: OrchestratorConfig,
    graph_backend: GraphBackend,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    llm: Any,
    catalog: ToolCatalog | None = None,
) -> MemoryOrchestrator:
    """Wire all internal components from configuration and backends.

    This is the primary DI entry point for library consumers.

    Args:
        config: Orchestrator configuration.
        graph_backend: Graph storage backend.
        vector_backend: Vector storage backend.
        embedding_provider: Text embedding provider.
        llm: LLM instance (BaseChatModel from langchain-core).
        catalog: Optional tool catalog. Creates empty one if not provided.

    Returns:
        Fully wired MemoryOrchestrator instance.
    """
    from symfonic.memory.orchestrator.factory import OrchestratorFactory

    return OrchestratorFactory.build(
        config=config,
        graph_backend=graph_backend,
        vector_backend=vector_backend,
        embedding_provider=embedding_provider,
        llm=llm,
        catalog=catalog,
    )

hydrate_context async

hydrate_context(
    scope: TenantScope,
    query: str,
    history: list[Any] | None = None,
) -> AssembledContext

Retrieve relevant memories and assemble context within budget.

.. deprecated:: hydrate_context is deprecated and will be removed in a future release. Hydration is now performed exclusively by SymfonicAgent._hydrate(), which iterates all enabled layers and runs spreading activation. This method is never called by the active pipeline.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
query	str	The user's current query.	required
history	list[Any] | None	Optional conversation history messages.	None

Returns:

Type	Description
AssembledContext	AssembledContext with system prompt, memories, tools, and history.

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

async def hydrate_context(
    self,
    scope: TenantScope,
    query: str,
    history: list[Any] | None = None,
) -> AssembledContext:
    """Retrieve relevant memories and assemble context within budget.

    .. deprecated::
        ``hydrate_context`` is deprecated and will be removed in a future
        release. Hydration is now performed exclusively by
        ``SymfonicAgent._hydrate()``, which iterates all enabled layers and
        runs spreading activation. This method is never called by the active
        pipeline.

    Args:
        scope: Tenant scope for isolation.
        query: The user's current query.
        history: Optional conversation history messages.

    Returns:
        AssembledContext with system prompt, memories, tools, and history.
    """
    import warnings

    warnings.warn(
        "MemoryOrchestrator.hydrate_context() is deprecated. "
        "Hydration is handled by SymfonicAgent._hydrate().",
        DeprecationWarning,
        stacklevel=2,
    )
    budget = self._config.context_budget
    history = history or []

    # Retrieve relevant memories across enabled layers.  ``top_k`` falls
    # back to the orchestrator's default top-k (the prior entry-count
    # field on ContextBudget was removed in v5.6.0 / v6.0 API freeze
    # prep).  This whole method is deprecated; the live pipeline is
    # ``SymfonicAgent._hydrate`` which uses ``max_tokens``.
    active_layers = set(self._config.enabled_layers) & set(self._layers.keys())
    entries = await self._retrieval.retrieve(
        scope=scope,
        query=query,
        layers=active_layers or None,
        top_k=self._config.default_top_k,
        embedding_provider=self._embedder,
    )

    # Route tools
    routing = await self._router.route(scope, query, history)

    # Assemble context
    bounded_history = history[-budget.max_history_messages:]
    memory_text = "\n".join(f"- [{e.layer.value}] {e.content}" for e in entries)
    graph_context = memory_text if entries else "No graph context available."
    retrieval_results = memory_text if entries else "No retrieval results."

    system_template = load_prompt("system")
    system_prompt = system_template.format(
        graph_context=graph_context,
        retrieval_results=retrieval_results,
    )

    return AssembledContext(
        system_prompt=system_prompt,
        memory_entries=entries,
        selected_tools=routing.selected_schemas[:budget.max_tools],
        conversation_history=bounded_history,
        total_estimated_tokens=len(system_prompt.split()) * 2,
    )

upgrade_durability async

upgrade_durability(
    scope: TenantScope,
    node_id: NodeId,
    new_durability: Durability,
) -> None

Flip the durability marker on a stored graph node (v7.26.2).

The explicit promotion / retirement signal for the consolidation gate. Typical use: write a transient row ("syncing..."), then once the state stabilises write a durable row ("complete: 42 messages") and call upgrade_durability(transient_id, "expired") so the transient row is pruned on the next Phase 8 cleanup.

Implementation (Contract E, verified per-backend at T0): all three graph backends' update_node either REPLACE the whole properties dict (InMemory, Postgres) or route a top-level properties key to a full $set (Mongo, after the v7.26.2 translation fix). A flat durability key cannot be used because it collides with the read-only MemoryNode.durability computed property. So this is a read-modify-write: fetch the node, mutate the properties dict, write the FULL dict back. This preserves every OTHER property on the node.

Raises:

Type	Description
ValueError	on an unknown durability literal (fail-loud).
KeyError / backend error	when the node does not exist.

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

async def upgrade_durability(
    self,
    scope: TenantScope,
    node_id: NodeId,
    new_durability: Durability,
) -> None:
    """Flip the durability marker on a stored graph node (v7.26.2).

    The explicit promotion / retirement signal for the consolidation
    gate.  Typical use: write a transient row ("syncing..."), then once
    the state stabilises write a durable row ("complete: 42 messages")
    and call ``upgrade_durability(transient_id, "expired")`` so the
    transient row is pruned on the next Phase 8 cleanup.

    Implementation (Contract E, verified per-backend at T0): all three
    graph backends' ``update_node`` either REPLACE the whole properties
    dict (InMemory, Postgres) or route a top-level ``properties`` key to
    a full ``$set`` (Mongo, after the v7.26.2 translation fix).  A flat
    ``durability`` key cannot be used because it collides with the
    read-only ``MemoryNode.durability`` computed property.  So this is a
    read-modify-write: fetch the node, mutate the properties dict, write
    the FULL dict back.  This preserves every OTHER property on the node.

    Raises:
        ValueError: on an unknown durability literal (fail-loud).
        KeyError / backend error: when the node does not exist.
    """
    validated = coerce_durability(new_durability)
    if self._graph is None:
        raise RuntimeError(
            "upgrade_durability requires a graph store; "
            "orchestrator was constructed without one.",
        )
    node = await self._graph.get_node(scope, node_id)
    if node is None:
        raise KeyError(
            f"Node {node_id} not found for tenant {scope.tenant_id}",
        )
    merged: dict[str, object] = dict(node.properties or {})
    merged["durability"] = validated
    # Full-dict write -- the backend full-replaces properties.  Because
    # we copied the existing dict first, no sibling property is lost.
    await self._graph.update_node(scope, node_id, {"properties": merged})
    logger.info(
        "Durability upgraded: tenant=%s node=%s -> %s",
        scope.tenant_id, node_id, validated,
    )

OrchestratorConfig

Bases: BaseModel

Immutable configuration for the MemoryOrchestrator and its components.

All tunable parameters are centralized here. Construct via with_defaults() for quick setup or from_env() to read from environment variables prefixed with SYMFONIC_GRAPH_MEMORY_.

from_env classmethod

from_env() -> OrchestratorConfig

Create a config by reading SYMFONIC_GRAPH_MEMORY_ environment variables.

v8.0: the three legacy env vars (SYMFONIC_GRAPH_MEMORY_COMPACTION_THRESHOLD, SYMFONIC_GRAPH_MEMORY_LLM_MODEL, SYMFONIC_GRAPH_MEMORY_LLM_TEMPERATURE) were silent no-ops with a DeprecationWarning for many minor releases. They now raise ValueError at config-load time so silent-setters get a loud signal instead of an invisible warning.

Supported variables

SYMFONIC_GRAPH_MEMORY_DEFAULT_TOP_K

Source code in src/symfonic/memory/orchestrator/config.py

@classmethod
def from_env(cls) -> OrchestratorConfig:
    """Create a config by reading SYMFONIC_GRAPH_MEMORY_ environment variables.

    v8.0: the three legacy env vars
    (``SYMFONIC_GRAPH_MEMORY_COMPACTION_THRESHOLD``,
    ``SYMFONIC_GRAPH_MEMORY_LLM_MODEL``,
    ``SYMFONIC_GRAPH_MEMORY_LLM_TEMPERATURE``) were silent no-ops
    with a ``DeprecationWarning`` for many minor releases.  They
    now raise ``ValueError`` at config-load time so silent-setters
    get a loud signal instead of an invisible warning.

    Supported variables:
        SYMFONIC_GRAPH_MEMORY_DEFAULT_TOP_K
    """
    kwargs: dict[str, object] = {}

    top_k = os.environ.get(f"{_ENV_PREFIX}DEFAULT_TOP_K")
    if top_k is not None:
        kwargs["default_top_k"] = int(top_k)

    # v8.0: loud failure for the three legacy no-op vars.
    _removed = {
        "COMPACTION_THRESHOLD": (
            "Use SleepConsolidator phases instead."
        ),
        "LLM_MODEL": (
            "LLM model is controlled by AgentConfig.model."
        ),
        "LLM_TEMPERATURE": (
            "LLM temperature is controlled by AgentConfig.model.temperature."
        ),
    }
    for suffix, guidance in _removed.items():
        full_name = f"{_ENV_PREFIX}{suffix}"
        if os.environ.get(full_name) is not None:
            raise ValueError(
                f"{full_name} was removed in v8.0 (it has been a "
                f"silent no-op since the v6.x era).  {guidance}  "
                f"Unset the variable in your deployment env."
            )

    return cls(**kwargs)

with_defaults classmethod

with_defaults() -> OrchestratorConfig

Create a config with all default values.

Source code in src/symfonic/memory/orchestrator/config.py

@classmethod
def with_defaults(cls) -> OrchestratorConfig:
    """Create a config with all default values."""
    return cls()

ProceduralLayer

ProceduralLayer(
    graph_store: GraphMemoryStore,
    dedup_threshold: float = 0.65,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
)

Procedural memory layer for skills and workflows.

BaseMemoryStore contract

• retrieve: queries skills by content matching
• write: stores a skill as a graph node
• summarize: returns empty string (skills are standalone)
• link: no-op (skills are standalone graph nodes)
• delete: removes a skill node from the graph

v7.3 Item 13.1: optional embedding_provider for auto-embed.

Default None preserves byte-identical pre-Item-13.1 behaviour. When supplied, every store_skill write runs through :func:maybe_embed before the de-dup janitor sees it -- so procedural skills auto-embed when the caller leaves node.embedding=None. Caller-provided embeddings always win. The embedding_cache is shared across all three graph-backed layers by the orchestrator.

Per-kind embedding-text rule: procedural skills always fall into the default branch of :func:symfonic.memory.embeddings.auto_embed.resolve_embedding_text and embed properties['content'] (the skill's rich description) rather than the label (which is the skill_id).

Source code in src/symfonic/memory/layers/procedural/layer.py

def __init__(
    self,
    graph_store: GraphMemoryStore,
    dedup_threshold: float = 0.65,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
) -> None:
    """v7.3 Item 13.1: optional ``embedding_provider`` for auto-embed.

    Default ``None`` preserves byte-identical pre-Item-13.1 behaviour.
    When supplied, every ``store_skill`` write runs through
    :func:`maybe_embed` before the de-dup janitor sees it -- so
    procedural skills auto-embed when the caller leaves
    ``node.embedding=None``. Caller-provided embeddings always win.
    The ``embedding_cache`` is shared across all three graph-backed
    layers by the orchestrator.

    Per-kind embedding-text rule: procedural skills always fall into
    the default branch of
    :func:`symfonic.memory.embeddings.auto_embed.resolve_embedding_text`
    and embed ``properties['content']`` (the skill's rich description)
    rather than the label (which is the skill_id).
    """
    self._graph = graph_store
    self._janitor = SemanticMerge(threshold=dedup_threshold)
    self._embedding_provider = embedding_provider
    self._embedding_cache = embedding_cache

approve_skill async

approve_skill(
    scope: TenantScope, node_id: NodeId
) -> MemoryNode

Promote a draft skill to approved+active.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def approve_skill(
    self, scope: TenantScope, node_id: NodeId,
) -> MemoryNode:
    """Promote a draft skill to approved+active."""
    node = await self._graph.get_node(scope, node_id)
    if node is None:
        msg = f"Procedural node {node_id} not found"
        raise KeyError(msg)
    merged = dict(node.properties or {})
    merged["status"] = "approved"
    merged["active"] = True
    return await self._graph.update_node(
        scope, node_id, {"properties": merged},
    )

delete async

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

Delete a skill node from the graph.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete a skill node from the graph."""
    await self._graph.delete_node(scope, node_id)

link async

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

No-op. Skills are standalone graph nodes.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def link(
    self, scope: TenantScope, source: NodeId, target: NodeId, relationship: str
) -> None:
    """No-op. Skills are standalone graph nodes."""
    return

query_skills async

query_skills(
    scope: TenantScope,
    query: str,
    top_k: int = 5,
    *,
    include_drafts: bool = False,
) -> list[MemoryEntry]

Query procedural skills by word-level matching.

Uses word intersection (like SemanticLayer.query_facts) instead of exact substring matching so that "deploy the app" can find a skill whose label/content contains "deploy" or "app".

By default only active (approved) skills are returned. Pass include_drafts=True to return skills of all statuses.

Ordering contract (v7.19.2): the returned list is sorted by node_id so the "first matching skill" downstream consumers rely on (precondition_gate's first-call wins, engine _maybe_resolve_forced_tool_choice's candidate procedure selection) is deterministic across the in-memory / mongodb / postgres backends. The in-memory backend's dict.values() order happens to be insertion-stable on CPython 3.7+, but mongodb / postgres do not guarantee registration order without an explicit sort. Sorting at query time pins the contract at the layer boundary so any future backend additions inherit it automatically.

v7.22 (T-7.22.12-15) multi-scope: when scope.inherits_from is non-empty, the query fans out to [scope, *scope.inherits_from] and merges results by shadow override on metadata['identifier'] first, then metadata['label']. Empty inherits_from (the common case) delegates to the single-scope fast path unchanged. See docs/concepts/procedural-shadow-inheritance.md for merge semantics and cache-invariant trade-offs.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def query_skills(
    self, scope: TenantScope, query: str, top_k: int = 5,
    *, include_drafts: bool = False,
) -> list[MemoryEntry]:
    """Query procedural skills by word-level matching.

    Uses word intersection (like SemanticLayer.query_facts) instead of
    exact substring matching so that "deploy the app" can find a skill
    whose label/content contains "deploy" or "app".

    By default only active (approved) skills are returned.  Pass
    ``include_drafts=True`` to return skills of all statuses.

    Ordering contract (v7.19.2): the returned list is sorted by
    ``node_id`` so the "first matching skill" downstream
    consumers rely on (precondition_gate's first-call wins,
    engine ``_maybe_resolve_forced_tool_choice``'s candidate
    procedure selection) is deterministic across the in-memory /
    mongodb / postgres backends.  The in-memory backend's
    ``dict.values()`` order happens to be insertion-stable on
    CPython 3.7+, but mongodb / postgres do not guarantee
    registration order without an explicit sort.  Sorting at
    query time pins the contract at the layer boundary so any
    future backend additions inherit it automatically.

    v7.22 (T-7.22.12-15) multi-scope: when ``scope.inherits_from``
    is non-empty, the query fans out to ``[scope, *scope.inherits_from]``
    and merges results by shadow override on
    ``metadata['identifier']`` first, then ``metadata['label']``.
    Empty ``inherits_from`` (the common case) delegates to the
    single-scope fast path unchanged.  See
    docs/concepts/procedural-shadow-inheritance.md for merge semantics
    and cache-invariant trade-offs.
    """
    # v8.0 subsumption (design §6.c): when the scope carries a multi-level
    # PATH, fan out over its ancestor prefixes (most-specific first) and
    # shadow-merge — the path model replaces the v7.22 inherits_from chain
    # for hierarchical procedural retrieval.  A legacy inherits_from chain
    # still works via _flatten_scope_chain (deprecated).  A pure 1-level
    # scope with no inherits_from keeps the byte-identical fast path.
    if scope.scope_depth > 1:
        scope_chain = self._path_scope_chain(scope)
    elif scope.inherits_from:
        scope_chain = self._flatten_scope_chain(scope)
    else:
        return await self._query_skills_single(
            scope, query, top_k, include_drafts=include_drafts,
        )
    # Per plan §3.e: each per-scope query asks for ``top_k *
    # len(scope_chain)`` headroom.  Worst-case fanout = ``top_k * 8``
    # entries scanned (chain capped by _MAX_INHERIT_DEPTH).  Acceptable
    # for procedural -- typically <100 entries per tenant.
    fanout = top_k * len(scope_chain)
    per_scope_results: list[list[MemoryEntry]] = []
    for single_scope in scope_chain:
        per_scope_results.append(
            await self._query_skills_single(
                single_scope, query, fanout,
                include_drafts=include_drafts,
            ),
        )
    return self._merge_with_shadow(per_scope_results, top_k=top_k)

reject_skill async

reject_skill(
    scope: TenantScope, node_id: NodeId
) -> MemoryNode

Mark a draft skill as rejected+inactive.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def reject_skill(
    self, scope: TenantScope, node_id: NodeId,
) -> MemoryNode:
    """Mark a draft skill as rejected+inactive."""
    node = await self._graph.get_node(scope, node_id)
    if node is None:
        msg = f"Procedural node {node_id} not found"
        raise KeyError(msg)
    merged = dict(node.properties or {})
    merged["status"] = "rejected"
    merged["active"] = False
    return await self._graph.update_node(
        scope, node_id, {"properties": merged},
    )

render_for_prompt

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

Render a procedural entry for MEMORY_CONTEXT.

v7.7: closes the v7.6.4 procedural-render-bypass structurally. When the entry carries v7.5+ skill-aware metadata (source starts with "authored:" OR non-empty precondition of either str or list[str] shape), delegate to the v7.5.1 _render_skill_for_prompt helper so PRE-FLIGHT markers reach the prompt. Pre-v7.5 procedural entries fall through to :func:render_legacy for byte-identical pre-v7.7 output and prompt-cache parity.

Supersedes the inline _is_skill_aware_entry predicate + dispatch block that lived in engine.py after the v7.6.4 hotfix; the engine's render loop is now polymorphic dispatch with zero procedural-specific knowledge.

Source code in src/symfonic/memory/layers/procedural/layer.py

def render_for_prompt(
    self,
    entry: Any,
    scrubber: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
) -> str:
    """Render a procedural entry for ``MEMORY_CONTEXT``.

    v7.7: closes the v7.6.4 procedural-render-bypass structurally.
    When the entry carries v7.5+ skill-aware metadata (``source``
    starts with ``"authored:"`` OR non-empty ``precondition`` of
    either ``str`` or ``list[str]`` shape), delegate to the
    v7.5.1 ``_render_skill_for_prompt`` helper so PRE-FLIGHT
    markers reach the prompt.  Pre-v7.5 procedural entries fall
    through to :func:`render_legacy` for byte-identical pre-v7.7
    output and prompt-cache parity.

    Supersedes the inline ``_is_skill_aware_entry`` predicate +
    dispatch block that lived in ``engine.py`` after the v7.6.4
    hotfix; the engine's render loop is now polymorphic dispatch
    with zero procedural-specific knowledge.
    """
    from symfonic.memory.layers.procedural.router import (
        _SkillRenderInput,
        render_skill_for_prompt,
    )

    metadata = getattr(entry, "metadata", {}) or {}
    source = metadata.get("source")
    is_authored = (
        isinstance(source, str) and source.startswith("authored:")
    )
    pre = metadata.get("precondition")
    has_pre = (
        (isinstance(pre, str) and pre.strip())
        or (
            isinstance(pre, list)
            and any(isinstance(p, str) and p.strip() for p in pre)
        )
    )
    if not (is_authored or has_pre):
        # Pre-v7.5 procedural -> legacy shape for cache parity.
        return render_legacy(entry, scrubber)

    # Defense-in-depth (Defect A) still runs on the skill-render path.
    content = strip_memory_label_prefix(
        str(getattr(entry, "content", "")),
    )
    rendered = render_skill_for_prompt(
        _SkillRenderInput(content=content, metadata=metadata),
    )
    layer_name = getattr(entry, "layer", "procedural")
    # ``_render_skill_for_prompt`` emits ``- <content> (steps: ...)``;
    # strip the leading bullet so the engine's ``[<layer>]`` prefix
    # produces the same shape as the v7.6.4 hotfix did.
    return f"[{layer_name}] " + rendered.lstrip("- ")

retrieve async

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

Retrieve skills matching the query.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def retrieve(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Retrieve skills matching the query."""
    return await self.query_skills(scope, query, top_k)

seed_authored_skill async

seed_authored_skill(
    scope: TenantScope,
    skill: MemoryEntry,
    *,
    source_tag: str,
    precondition: str
    | list[str | dict[str, Any]]
    | None = None,
) -> MemoryNode

Seed a hand-authored procedural skill (Jarvio Ask 7 Option B).

Convenience wrapper around :meth:store_skill for the domain-plugin tier of procedural memory. The result lands at status="approved" (immediately active in :meth:query_skills), carries the required source tag so :class:~symfonic.memory.janitor.SemanticMerge leaves it alone, and optionally records one or more pre-flight preconditions.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
skill	MemoryEntry	The skill MemoryEntry to seed -- domain plugins build this with the label, content, steps and context they author. Any existing metadata['source'] / metadata['precondition'] keys are overwritten by the explicit source_tag / precondition kwargs.	required
source_tag	str	A discriminator the janitor uses to recognise the authored tier. MUST start with "authored:" -- convention is "authored:<plugin-name>" so multiple domain plugins can coexist without colliding on a shared bucket. Anything else raises ValueError.	required
precondition	str | list[str | dict[str, Any]] | None	Optional pre-flight rule. Three shapes accepted (v7.6.2 -- Jarvio multi-precondition ask): None -- no precondition stored. str -- single free-text rule (v7.5.0 shape, byte-identical to pre-7.6.2 behaviour when passed). list[str] -- ordered list of pre-flight rules. Empty strings are stripped; an all-empty list is treated as None. A 1-element list is collapsed back to str so the prompt-cache stays warm for callers who pass single-element lists. The list semantic is "in order, ALL pre-flights MUST fire before the action tool" -- the v7.5.1 CapabilityRouter renders one indented PRE-FLIGHT: line per entry, in supplied order, and the prompt clause at prompts/router.txt characterises the ordering as a hard constraint.	None

Returns:

Type	Description
MemoryNode	The persisted :class:MemoryNode.

Raises:

Type	Description
ValueError	When source_tag does not start with the authored prefix. Fail-fast so the janitor's guard is actually engaged for the resulting node.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def seed_authored_skill(
    self,
    scope: TenantScope,
    skill: MemoryEntry,
    *,
    source_tag: str,
    precondition: str | list[str | dict[str, Any]] | None = None,
) -> MemoryNode:
    """Seed a hand-authored procedural skill (Jarvio Ask 7 Option B).

    Convenience wrapper around :meth:`store_skill` for the
    domain-plugin tier of procedural memory.  The result lands at
    ``status="approved"`` (immediately active in
    :meth:`query_skills`), carries the required ``source`` tag so
    :class:`~symfonic.memory.janitor.SemanticMerge` leaves it
    alone, and optionally records one or more pre-flight
    preconditions.

    Args:
        scope: Tenant scope for isolation.
        skill: The skill ``MemoryEntry`` to seed -- domain plugins
            build this with the label, content, steps and context
            they author.  Any existing ``metadata['source']`` /
            ``metadata['precondition']`` keys are overwritten by
            the explicit ``source_tag`` / ``precondition`` kwargs.
        source_tag: A discriminator the janitor uses to recognise
            the authored tier.  MUST start with ``"authored:"`` --
            convention is ``"authored:<plugin-name>"`` so multiple
            domain plugins can coexist without colliding on a
            shared bucket.  Anything else raises ``ValueError``.
        precondition: Optional pre-flight rule.  Three shapes
            accepted (v7.6.2 -- Jarvio multi-precondition ask):

            * ``None`` -- no precondition stored.
            * ``str`` -- single free-text rule (v7.5.0 shape,
              byte-identical to pre-7.6.2 behaviour when passed).
            * ``list[str]`` -- ordered list of pre-flight rules.
              Empty strings are stripped; an all-empty list is
              treated as ``None``.  A 1-element list is collapsed
              back to ``str`` so the prompt-cache stays warm for
              callers who pass single-element lists.  The list
              semantic is "in order, ALL pre-flights MUST fire
              before the action tool" -- the v7.5.1
              ``CapabilityRouter`` renders one indented
              ``PRE-FLIGHT:`` line per entry, in supplied order,
              and the prompt clause at
              ``prompts/router.txt`` characterises the ordering as
              a hard constraint.

    Returns:
        The persisted :class:`MemoryNode`.

    Raises:
        ValueError: When ``source_tag`` does not start with the
            authored prefix.  Fail-fast so the janitor's guard is
            actually engaged for the resulting node.
    """
    if not source_tag.startswith(SemanticMerge._AUTHORED_SOURCE_PREFIX):
        raise ValueError(
            "source_tag must start with "
            f"'{SemanticMerge._AUTHORED_SOURCE_PREFIX}'; "
            f"got {source_tag!r}",
        )
    metadata: dict[str, object] = {
        **(skill.metadata or {}),
        "source": source_tag,
    }
    normalised = _normalise_precondition(precondition)
    if normalised is not None:
        metadata["precondition"] = normalised
    # ``MemoryEntry`` is a pydantic model -- ``model_copy`` keeps the
    # caller's instance untouched so the helper has no aliasing
    # surprises if the same entry is reused across calls.
    authored = skill.model_copy(update={"metadata": metadata})
    return await self.store_skill(scope, authored, status="approved")

store_skill async

store_skill(
    scope: TenantScope,
    skill: MemoryEntry,
    *,
    status: str = "draft",
) -> MemoryNode

Store a skill as a MemoryNode with steps and context properties.

Before inserting, the incoming skill is checked against existing procedural nodes via SemanticMerge. If a sufficiently similar node already exists the two are merged in-place and no new node is created.

The skill's metadata should include 'steps' (list[str]) and 'context' (str) keys for proper procedural storage.

Parameters:

Name	Type	Description	Default
status	str	"draft" (inactive, pending review), "approved" (active), "rejected" (archived, will not be queried).	'draft'

Source code in src/symfonic/memory/layers/procedural/layer.py

async def store_skill(
    self, scope: TenantScope, skill: MemoryEntry,
    *, status: str = "draft",
) -> MemoryNode:
    """Store a skill as a MemoryNode with steps and context properties.

    Before inserting, the incoming skill is checked against existing
    procedural nodes via SemanticMerge.  If a sufficiently similar node
    already exists the two are merged in-place and no new node is created.

    The skill's metadata should include 'steps' (list[str]) and
    'context' (str) keys for proper procedural storage.

    Args:
        status: "draft" (inactive, pending review), "approved" (active),
                "rejected" (archived, will not be queried).
    """
    # Prefer metadata label (set by engine) over content truncation
    # so procedural labels show skill_id, not truncated JSON.
    node_label = skill.metadata.get("label") or skill.content[:100]
    node = MemoryNode(
        layer=MemoryLayer.PROCEDURAL,
        tenant_id=scope.tenant_id,
        label=str(node_label)[:100],
        properties={
            "content": skill.content,
            "steps": skill.metadata.get("steps", []),
            "context": skill.metadata.get("context", ""),
            "status": status,
            "active": status == "approved",
            **{
                k: v for k, v in skill.metadata.items()
                if k not in ("steps", "context", "status", "active")
            },
        },
        importance=skill.importance,
    )
    if skill.node_id:
        node.id = skill.node_id

    # v7.3 Item 13.1: auto-embed before the de-dup janitor so the
    # canonical node carries an embedding when the operator opted into
    # an EmbeddingProvider. Caller-provided embeddings always win;
    # provider failures degrade to embedding=None (see
    # auto_embed.maybe_embed docstring).
    # TODO(v7.4): batch embeddings when called in a tight loop.
    node = await maybe_embed(
        node, self._embedding_provider, self._embedding_cache,
    )

    canonical = await self._janitor.deduplicate(scope, self._graph, node)
    if canonical.id != node.id:
        # A merge happened; return the updated canonical node without
        # adding a duplicate to the graph.
        return canonical

    return await self._graph.add_node(scope, node)

summarize async

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

Not meaningful for procedural skills. Returns empty string.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def summarize(self, scope: TenantScope, entries: list[MemoryEntry]) -> str:
    """Not meaningful for procedural skills. Returns empty string."""
    return ""

write async

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

Write a skill entry to the graph store.

Source code in src/symfonic/memory/layers/procedural/layer.py

async def write(self, scope: TenantScope, entry: MemoryEntry) -> None:
    """Write a skill entry to the graph store."""
    await self.store_skill(scope, entry)

ScoringWeights

Bases: BaseModel

Weights for the four-signal retrieval scoring formula.

The four weights must sum to 1.0 (within floating-point tolerance). Default allocation: semantic 40%, recency 20%, frequency 20%, graph 20%.

TenantScope

Bases: BaseModel

Immutable hierarchical N-level multi-tenant request scope.

Canonical representation is the root-first path (path[0] broadest, path[-1] most-specific). The legacy tenant_id / sub_tenant_id fields are RETAINED for backward compatibility and, when no explicit path override is supplied, ARE the 1-/2-level path. When an explicit N-level path is supplied via the factories, tenant_id / sub_tenant_id read back off the path.

Isolation (design §5.d): a query at path P sees a memory at path Q iff Q is a prefix of P. Enforced via :meth:ancestor_prefix_paths exact-IN at the backend, NEVER a string prefix scan.

path property

path: tuple[ScopeLevel, ...]

Root-first ordered path (canonical hierarchical representation).

Derived from the flat fields when no explicit override is set

• tenant_id only → 1-level (tenant:tenant_id,)
• • sub_tenant_id → 2-level (tenant:tenant_id, sub_tenant:sub)

scope_depth property

scope_depth: int

Number of levels in the path (1-based; root-only path is depth 1).

scope_path property

scope_path: str

Materialised US-delimited, root-first path string for the backend.

org\x1facme\x1fbrand\x1fmyhalos\x1fconversation\x1fc-123. This is the stored column/field value (design §5.b).

ancestor_prefix_paths

ancestor_prefix_paths() -> list[str]

The exact-IN list for the isolation filter (design §5.b/§5.d).

Returns the materialised scope_path string of EVERY prefix of this scope's path (root → … → self), inclusive. A query at path P uses this list as WHERE scope_path = ANY(list) so it matches a stored memory at path Q iff Q is a prefix of P — exact set membership, never a LIKE 'prefix%' scan (which would leak acme-corp to acme).

Source code in src/symfonic/core/scope.py

def ancestor_prefix_paths(self) -> list[str]:
    """The exact-IN list for the isolation filter (design §5.b/§5.d).

    Returns the materialised ``scope_path`` string of EVERY prefix of this
    scope's path (root → … → self), inclusive.  A query at path P uses
    this list as ``WHERE scope_path = ANY(list)`` so it matches a stored
    memory at path Q iff Q is a prefix of P — exact set membership, never
    a ``LIKE 'prefix%'`` scan (which would leak ``acme-corp`` to ``acme``).
    """
    levels = self.path
    return [
        scope_path_for(levels[: i + 1]) for i in range(len(levels))
    ]

child

child(kind: str, id: str) -> TenantScope

Return a new scope with one more (deeper) level appended.

org_scope.child("brand", "myhalos").child("conversation", "c-123") builds the full hierarchy. Namespace and inherits_from are preserved.

The result preserves the receiver's concrete type (type(self)) so chaining .child() off a subclass such as :class:~symfonic.agent.types.FrameworkTenantScope keeps the subclass (and its converter methods) instead of silently downcasting to the base TenantScope (P1 — subclass-drop on chaining).

Source code in src/symfonic/core/scope.py

def child(self, kind: str, id: str) -> TenantScope:
    """Return a new scope with one more (deeper) level appended.

    ``org_scope.child("brand", "myhalos").child("conversation", "c-123")``
    builds the full hierarchy.  Namespace and inherits_from are preserved.

    The result preserves the receiver's concrete type (``type(self)``) so
    chaining ``.child()`` off a subclass such as
    :class:`~symfonic.agent.types.FrameworkTenantScope` keeps the subclass
    (and its converter methods) instead of silently downcasting to the
    base ``TenantScope`` (P1 — subclass-drop on chaining).
    """
    new_path = (*self.path, ScopeLevel(kind=kind, id=id))
    return type(self)(
        tenant_id=new_path[0].id,
        sub_tenant_id=new_path[1].id if len(new_path) > 1 else None,
        namespace=self.namespace,
        inherits_from=self.inherits_from,
        path_override=new_path,
    )

from_path classmethod

from_path(
    path: list[ScopeLevel] | tuple[ScopeLevel, ...],
    *,
    namespace: str | None = None,
) -> TenantScope

Construct an N-level scope from an explicit root-first path.

Source code in src/symfonic/core/scope.py

@classmethod
def from_path(
    cls,
    path: list[ScopeLevel] | tuple[ScopeLevel, ...],
    *,
    namespace: str | None = None,
) -> TenantScope:
    """Construct an N-level scope from an explicit root-first path."""
    levels = tuple(path)
    if not levels:
        raise ValueError("TenantScope.from_path requires a non-empty path")
    tenant_id = levels[0].id
    sub_tenant_id = levels[1].id if len(levels) > 1 else None
    return cls(
        tenant_id=tenant_id,
        sub_tenant_id=sub_tenant_id,
        namespace=namespace,
        path_override=levels,
    )

from_request classmethod

from_request(
    tenant_id: str,
    *,
    sub_tenant_id: str | None = None,
    namespace: str | None = None,
    path: list[ScopeLevel]
    | tuple[ScopeLevel, ...]
    | None = None,
) -> TenantScope

Construct a scope from request-edge fields (blessed adopter entry).

Supply path for N-level hierarchies, or the flat fields for the legacy 1-/2-level shape.

Source code in src/symfonic/core/scope.py

@classmethod
def from_request(
    cls,
    tenant_id: str,
    *,
    sub_tenant_id: str | None = None,
    namespace: str | None = None,
    path: list[ScopeLevel] | tuple[ScopeLevel, ...] | None = None,
) -> TenantScope:
    """Construct a scope from request-edge fields (blessed adopter entry).

    Supply ``path`` for N-level hierarchies, or the flat fields for the
    legacy 1-/2-level shape.
    """
    if path is not None:
        return cls.from_path(path, namespace=namespace)
    return cls(
        tenant_id=tenant_id,
        sub_tenant_id=sub_tenant_id,
        namespace=namespace,
    )

from_state_dict classmethod

from_state_dict(state: Mapping[str, Any]) -> TenantScope

Reconstruct a :class:TenantScope from a state-dict mapping.

Accepts UPPER_SNAKE and lower_snake casings (UPPER wins). Reads an explicit path / PATH key (list of {kind, id} dicts) when present — the N-level form; otherwise derives from the legacy flat fields. inherits_from is read recursively (v7.22 compat).

Source code in src/symfonic/core/scope.py

@classmethod
def from_state_dict(cls, state: Mapping[str, Any]) -> TenantScope:
    """Reconstruct a :class:`TenantScope` from a state-dict mapping.

    Accepts UPPER_SNAKE and lower_snake casings (UPPER wins).  Reads an
    explicit ``path`` / ``PATH`` key (list of ``{kind, id}`` dicts) when
    present — the N-level form; otherwise derives from the legacy flat
    fields.  ``inherits_from`` is read recursively (v7.22 compat).
    """
    def _read(*keys: str) -> Any:
        for k in keys:
            if k in state and state[k] is not None:
                return state[k]
        return None

    raw_path = _read("PATH", "path")
    if isinstance(raw_path, (list, tuple)) and raw_path:
        # Fail loud on malformed levels rather than silently dropping
        # them (silent scope data loss). Round-2 fix (triage 2026-07-02).
        from symfonic.agent.types import ScopeValidationError

        built: list[ScopeLevel] = []
        for lvl in raw_path:
            if not (isinstance(lvl, Mapping) and "kind" in lvl and "id" in lvl):
                raise ScopeValidationError(
                    "Malformed scope path level in from_state_dict: "
                    f"{lvl!r} (each level requires 'kind' and 'id').",
                )
            built.append(ScopeLevel(kind=str(lvl["kind"]), id=str(lvl["id"])))
        levels = tuple(built)
        if levels:
            namespace = _read("NAMESPACE", "namespace")
            return cls.from_path(levels, namespace=namespace)

    tenant_id = _read("TENANT_ID", "tenant_id") or ""
    sub_tenant_id = _read("SUB_TENANT_ID", "sub_tenant_id")
    namespace = _read("NAMESPACE", "namespace")

    raw_inherits = _read("INHERITS_FROM", "inherits_from")
    inherits_from: tuple[TenantScope, ...] = ()
    if isinstance(raw_inherits, (list, tuple)) and raw_inherits:
        inherits_from = tuple(
            cls.from_state_dict(ancestor)
            for ancestor in raw_inherits
            if isinstance(ancestor, Mapping)
        )

    kwargs: dict[str, Any] = {"tenant_id": tenant_id}
    if sub_tenant_id is not None:
        kwargs["sub_tenant_id"] = sub_tenant_id
    if namespace is not None:
        kwargs["namespace"] = namespace
    if inherits_from:
        kwargs["inherits_from"] = inherits_from
    return cls(**kwargs)

narrow

narrow(sub_tenant_id: str) -> TenantScope

Create a new scope narrowed to a sub-tenant (legacy 2-level shim).

Retained byte-identically from v7.22: sets both sub_tenant_id and namespace to the supplied value and preserves inherits_from. For N-level hierarchies prefer :meth:child.

Source code in src/symfonic/core/scope.py

def narrow(self, sub_tenant_id: str) -> TenantScope:
    """Create a new scope narrowed to a sub-tenant (legacy 2-level shim).

    Retained byte-identically from v7.22: sets both ``sub_tenant_id`` and
    ``namespace`` to the supplied value and preserves ``inherits_from``.
    For N-level hierarchies prefer :meth:`child`.
    """
    return TenantScope(
        tenant_id=self.tenant_id,
        sub_tenant_id=sub_tenant_id,
        namespace=sub_tenant_id,
        inherits_from=self.inherits_from,
    )

root classmethod

root(
    kind: str, id: str, *, namespace: str | None = None
) -> TenantScope

Construct a 1-level root scope (kind:id,).

Source code in src/symfonic/core/scope.py

@classmethod
def root(cls, kind: str, id: str, *, namespace: str | None = None) -> TenantScope:
    """Construct a 1-level root scope ``(kind:id,)``."""
    return cls.from_path([ScopeLevel(kind=kind, id=id)], namespace=namespace)

to_state_dict

to_state_dict() -> dict[str, Any]

Convert to state-dict fields for BaseAgentState injection.

Backward-compatible: a legacy 1-/2-level scope (no explicit path override) emits EXACTLY the pre-v8.0 keys (tenant_id + sub_tenant_id [+ namespace] [+ inherits_from]) — byte-identical. An explicit N-level path additionally emits a path key (list of {kind, id} dicts) so the hierarchy round-trips.

Source code in src/symfonic/core/scope.py

def to_state_dict(self) -> dict[str, Any]:
    """Convert to state-dict fields for BaseAgentState injection.

    Backward-compatible: a legacy 1-/2-level scope (no explicit path
    override) emits EXACTLY the pre-v8.0 keys
    (``tenant_id`` + ``sub_tenant_id`` [+ ``namespace``] [+
    ``inherits_from``]) — byte-identical.  An explicit N-level path
    additionally emits a ``path`` key (list of ``{kind, id}`` dicts) so
    the hierarchy round-trips.
    """
    d: dict[str, Any] = {
        "tenant_id": self.tenant_id,
        "sub_tenant_id": self.sub_tenant_id,
    }
    if self.namespace is not None:
        d["namespace"] = self.namespace
    if self.inherits_from:
        d["inherits_from"] = [
            ancestor.to_state_dict() for ancestor in self.inherits_from
        ]
    if self.path_override is not None:
        d["path"] = [
            {"kind": lvl.kind, "id": lvl.id} for lvl in self.path_override
        ]
    return d

ToolCatalog

ToolCatalog()

Registry of available tools and capabilities.

Tools are registered dynamically and can be searched by keyword. The catalog provides tool metadata to the CapabilityRouter for LLM-driven tool selection.

Source code in src/symfonic/memory/layers/procedural/catalog.py

def __init__(self) -> None:
    self._tools: dict[str, dict[str, Any]] = {}

get

get(tool_id: str) -> dict[str, Any] | None

Get a tool by its ID. Returns None if not found.

Source code in src/symfonic/memory/layers/procedural/catalog.py

def get(self, tool_id: str) -> dict[str, Any] | None:
    """Get a tool by its ID. Returns None if not found."""
    return self._tools.get(tool_id)

list_all

list_all() -> list[dict[str, Any]]

Return all registered tools.

Source code in src/symfonic/memory/layers/procedural/catalog.py

def list_all(self) -> list[dict[str, Any]]:
    """Return all registered tools."""
    return list(self._tools.values())

register

register(
    tool_id: str, description: str, schema: dict[str, Any]
) -> None

Register a tool with its description and schema.

Parameters:

Name	Type	Description	Default
tool_id	str	Unique identifier for the tool.	required
description	str	Human-readable description of what the tool does.	required
schema	dict[str, Any]	JSON schema or dict describing the tool's parameters.	required

Source code in src/symfonic/memory/layers/procedural/catalog.py

def register(self, tool_id: str, description: str, schema: dict[str, Any]) -> None:
    """Register a tool with its description and schema.

    Args:
        tool_id: Unique identifier for the tool.
        description: Human-readable description of what the tool does.
        schema: JSON schema or dict describing the tool's parameters.
    """
    self._tools[tool_id] = {
        "tool_id": tool_id,
        "description": description,
        "schema": schema,
    }

search

search(query: str) -> list[dict[str, Any]]

Search tools by keyword matching against ID and description.

Simple case-insensitive substring matching for v1.

Source code in src/symfonic/memory/layers/procedural/catalog.py

def search(self, query: str) -> list[dict[str, Any]]:
    """Search tools by keyword matching against ID and description.

    Simple case-insensitive substring matching for v1.
    """
    query_lower = query.lower()
    results: list[dict[str, Any]] = []
    for tool in self._tools.values():
        tool_id_lower = tool["tool_id"].lower()
        desc_lower = tool["description"].lower()
        if query_lower in tool_id_lower or query_lower in desc_lower:
            results.append(tool)
    return results

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

WritePolicy

Bases: BaseModel

Controls which operations are applied and how many can be pending.

Operations with importance below min_importance_threshold are silently dropped during commit_pending.