Skip to content

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 
 98
 99
100
101
102
103
104
105
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 
129
130
131
132
133
134
135
136
137
138
139
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 
92
93
94
95
96
97
async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete a memory entry/node from this layer.

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

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

Create a relationship link between two nodes.

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

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

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

render_for_prompt 

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

Render a single retrieved entry for the MEMORY_CONTEXT block.

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

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

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

Source code in src/symfonic/memory/protocols.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def render_for_prompt(
    self,
    entry: Any,
    scrubber: Callable[[dict[str, Any]], dict[str, Any]] | None = None,
) -> str:
    """Render a single retrieved entry for the ``MEMORY_CONTEXT`` block.

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

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

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

retrieve async 

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

Retrieve memory entries matching the query.

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

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

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

summarize async 

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

Produce a summary of the given entries.

Returns an empty string if summarization is not meaningful.

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

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

write async 

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

Write a memory entry to this layer.

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

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

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

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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
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 (adopter 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 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
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 (adopter 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 adopter 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 
27
28
29
30
31
32
33
34
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)

framework_default classmethod 

framework_default() -> ContextBudget

The framework's own default, built without the adopter-facing warning.

OrchestratorConfig.context_budget needs an instance, and taking it from __init__ made the framework fire a deprecation about its own default on the first line of every platform program — pinned as EX-3 in the warning budget and recorded there as "not actionable by an adopter", because there was nothing for them to change.

A deprecation warns whoever chose the deprecated thing. Nobody chose this one, so nobody is warned for it. An adopter who constructs ContextBudget() is still warned, and a test pins both halves.

model_construct rather than a suppressed __init__: suppression needs warnings.catch_warnings, which mutates process-global filter state and would briefly swallow another thread's warnings. Validation loses nothing here — every field is a literal default this class declares, with no input to check.

Source code in src/symfonic/memory/models/context.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
@classmethod
def framework_default(cls) -> ContextBudget:
    """The framework's own default, built without the adopter-facing warning.

    ``OrchestratorConfig.context_budget`` needs an instance, and taking it
    from ``__init__`` made the framework fire a deprecation *about its own
    default* on the first line of every platform program — pinned as EX-3 in
    the warning budget and recorded there as "not actionable by an adopter",
    because there was nothing for them to change.

    A deprecation warns whoever chose the deprecated thing. Nobody chose
    this one, so nobody is warned for it. An adopter who constructs
    ``ContextBudget()`` **is** still warned, and a test pins both halves.

    ``model_construct`` rather than a suppressed ``__init__``: suppression
    needs ``warnings.catch_warnings``, which mutates process-global filter
    state and would briefly swallow another thread's warnings. Validation
    loses nothing here — every field is a literal default this class
    declares, with no input to check.
    """
    return cls.model_construct()

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 
132
133
134
async def embed(self, text: str) -> list[float]:
    """Embed a single text string into a vector."""
    ...

embed_batch async 

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

Embed multiple texts into vectors.

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

GraphBackend

Bases: Protocol

Protocol for graph node and edge persistence.

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

add_edge async 

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

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

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

add_node async 

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

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

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

delete_edge async 

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

Delete a single edge by its ID.

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

delete_node async 

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

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

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

delete_subtree async 

delete_subtree(scope: TenantScope) -> int

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

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

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

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

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

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

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

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

get_neighbors async 

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

Get neighboring nodes, optionally filtered by relationship type.

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

get_node async 

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

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

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

query_edges async 

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

Query edges with optional filters and pagination.

Parameters:

Name Type Description Default
scope TenantScope

Tenant isolation scope.

 required
filters dict[str, Any] | None

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

 None
limit int

Maximum number of edges to return.

 50
offset int

Number of edges to skip (for pagination).

 0

Returns:

Type Description
list[Any]

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

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

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

query_nodes async 

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

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

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

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

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

query_subtree async 

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

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

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

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

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

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

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

traverse async 

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

Traverse the graph from a start node up to max_depth.

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

update_node async 

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

Partial update of node properties. Returns the updated node.

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

upsert_edge async 

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

Insert edge or increment weight if it already exists.

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

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

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

GraphMemoryStore 

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

Bases: GraphQueryMixin

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 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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

backend property 

backend: GraphBackend

What this store persists through. Named, so callers stop reaching.

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 
86
87
88
89
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 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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 
279
280
281
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 
218
219
220
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)

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 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
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 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
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)

neighbor_probe async 

neighbor_probe(scope: TenantScope, node_id: NodeId, minimum: int = 1, relationship: str | None = None, target: str | None = None) -> bool

Answer a threshold/existence question without loading a neighborhood.

Source code in src/symfonic/memory/graph/store.py 
171
172
173
174
175
176
177
178
async def neighbor_probe(self, scope: TenantScope, node_id: NodeId, minimum: int = 1,
                         relationship: str | None = None, target: str | None = None) -> bool:
    """Answer a threshold/existence question without loading a neighborhood."""
    probe = getattr(self._backend, "neighbor_probe", None)
    if callable(probe):
        return bool(await probe(scope, str(node_id), minimum, relationship, target))
    neighbors = await self.get_neighbors(scope, node_id, relationship)
    return sum(target is None or str(node.id) == target for node in neighbors) >= minimum

related_candidates async 

related_candidates(scope: TenantScope, cue: str) -> list[MemoryNode]

Bounded exact-owner semantic candidates for offline comparison.

Source code in src/symfonic/memory/graph/store.py 
180
181
182
183
184
185
186
187
188
189
async def related_candidates(self, scope: TenantScope, cue: str) -> list[MemoryNode]:
    """Bounded exact-owner semantic candidates for offline comparison."""
    select = getattr(self._backend, "select_candidates", None)
    if not callable(select):
        return []
    from symfonic.memory.candidates import CandidateRequest
    page = await select(scope, CandidateRequest(
        cue=cue, layers=("semantic",), limit=64, profile_slots=0, exact_scope=True,
    ))
    return list(page.nodes)

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 
212
213
214
215
216
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 
91
92
93
94
95
96
97
98
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 
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
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()

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

Testing and local development only.

Source code in src/symfonic/memory/backends/in_memory.py 
43
44
45
46
def __init__(self) -> None:
    self._nodes: dict[str, dict[str, MemoryNode]] = defaultdict(dict)
    self._edges: dict[str, dict[str, MemoryEdge]] = defaultdict(dict)
    self._workset_cursors: dict[Any, Any] = {}

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 
111
112
113
114
115
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 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
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 
239
240
241
242
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 
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
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-gated by prefix visibility like the PG delete_node and
    # the get_node read filter, so a sibling sub-scope cannot delete this.
    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]

delete_subtree async 

delete_subtree(scope: TenantScope) -> int

Erase owned subtree, edges and cursors in one non-yielding operation.

Source code in src/symfonic/memory/backends/in_memory.py 
244
245
246
247
async def delete_subtree(self, scope: TenantScope) -> int:
    """Erase owned subtree, edges and cursors in one non-yielding operation."""
    from symfonic.memory.backends.in_memory_deletion import delete_subtree
    return delete_subtree(self, scope)

get_neighbors async 

get_neighbors(scope: 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 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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 
117
118
119
120
121
122
123
124
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 
264
265
266
267
268
269
270
271
272
273
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."""
    from symfonic.memory.backends.in_memory_queries import query_edges
    return query_edges(self._edges.get(scope.tenant_id, {}).values(), filters, limit, offset)

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 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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

query_subtree async 

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

Filter nodes at scope or below it (the descendant read).

Source code in src/symfonic/memory/backends/in_memory.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def query_subtree(
    self, scope: TenantScope, filters: dict[str, Any], limit: int | None = None
) -> list[MemoryNode]:
    """Filter nodes at ``scope`` or below it (the descendant read)."""
    if limit == 0:
        return []
    root = materialise_scope_path(scope)
    results: list[MemoryNode] = []
    for node in self._nodes.get(scope.tenant_id, {}).values():
        stored = stored_scope_path(node.properties, node.tenant_id)
        if not is_in_subtree(stored, root):
            continue
        if self._matches_filters(node, filters):
            results.append(node)
            if limit is not None and len(results) >= limit:
                break
    return results

transaction 

transaction() -> Any

All of a batch, or none of it -- see :mod:.in_memory_transaction.

Source code in src/symfonic/memory/backends/in_memory.py 
38
39
40
41
def transaction(self) -> Any:
    """All of a batch, or none of it -- see :mod:`.in_memory_transaction`."""
    from symfonic.memory.backends.in_memory_transaction import snapshot
    return snapshot(self)

traverse async 

traverse(scope: 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 
149
150
151
152
153
154
async def traverse(
    self, scope: TenantScope, start: NodeId, max_depth: int
) -> list[MemoryNode]:
    """BFS traversal from start node up to max_depth."""
    from symfonic.memory.backends.in_memory_queries import traverse
    return await traverse(self, scope, start, max_depth)

update_node async 

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

Partial update of node properties.

Source code in src/symfonic/memory/backends/in_memory.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
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 
249
250
251
252
253
254
255
256
257
258
259
260
261
262
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_vector.py 
49
50
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_vector.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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_vector.py 
122
123
124
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_vector.py 
105
106
107
108
109
110
111
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_vector.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
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 
103
104
105
106
107
108
109
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 
 99
100
101
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)

Bases: OrchestratorWriteMixin

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 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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()

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 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
@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 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
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,
    )

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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@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 
104
105
106
107
@classmethod
def with_defaults(cls) -> OrchestratorConfig:
    """Create a config with all default values."""
    return cls()

PredicateOutcome dataclass 

PredicateOutcome(matched: bool, reason: str)

Whether the predicate matched, and a line saying why.

reason names the path, the operator, the value the rule declared, and how the turn's state compared -- never what the state held.

That last exclusion is a correction. The first version echoed the resolved value, reasoning that a decision reason is not an audit record. It is worse than one: filter_skills_by_state_predicates writes this string to a DEBUG log on the legacy path, so an entitlement, an account id or anything else a deployment puts in turn state was landing in application logs by default. "region eq 'US' -> False (state: differs)" is as actionable as naming the value and names nothing that belongs to the turn.

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 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
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 
737
738
739
740
741
742
743
744
745
746
747
748
749
750
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 
256
257
258
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(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 
250
251
252
253
254
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 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
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 
752
753
754
755
756
757
758
759
760
761
762
763
764
765
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 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
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 
236
237
238
239
240
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 (adopter 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 -- adopter 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 
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
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 (adopter 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 -- adopter 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 
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
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 
246
247
248
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 
242
243
244
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\x1fwidgets\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 
340
341
342
343
344
345
346
347
348
349
350
351
352
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", "widgets").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 
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
def child(self, kind: str, id: str) -> TenantScope:
    """Return a new scope with one more (deeper) level appended.

    ``org_scope.child("brand", "widgets").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 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
@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 
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
@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 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
@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.errors 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 
402
403
404
405
406
407
408
409
410
411
412
413
414
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 
376
377
378
379
@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 
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
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 
20
21
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 
37
38
39
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 
41
42
43
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 
23
24
25
26
27
28
29
30
31
32
33
34
35
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 
45
46
47
48
49
50
51
52
53
54
55
56
57
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 
145
146
147
148
149
150
151
152
153
154
async def add(
    self,
    scope: TenantScope,
    ids: list[str],
    embeddings: list[list[float]],
    metadatas: list[dict[str, Any]],
    documents: list[str],
) -> None:
    """Add vectors with metadata to the store."""
    ...

count async 

count(scope: TenantScope) -> int

Return the total number of stored vectors for the tenant.

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

delete async 

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

Delete vectors by their IDs.

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

delete_subtree async 

delete_subtree(scope: TenantScope) -> int

Delete every vector at scope and below it.

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

search async 

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

Search for the top_k most similar vectors.

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

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

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

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.

evaluate_state_predicate 

evaluate_state_predicate(state: Any, predicate: Any) -> PredicateOutcome

Answer one {"path": ..., "op": ..., "value": ...} against state.

Source code in src/symfonic/memory/layers/procedural/predicates.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def evaluate_state_predicate(state: Any, predicate: Any) -> PredicateOutcome:
    """Answer one ``{"path": ..., "op": ..., "value": ...}`` against ``state``."""
    if not isinstance(predicate, dict):
        return PredicateOutcome(False, "non-dict predicate")
    path, op = predicate.get("path"), predicate.get("op")
    if not isinstance(path, str) or not path:
        return PredicateOutcome(False, "missing or empty `path`")
    if not isinstance(op, str) or op not in ALLOWED_OPS:
        return PredicateOutcome(
            False, f"unknown operator `{op!r}` (allowed: {ALLOWED_OPS})"
        )
    resolved = resolve_path(state, path)

    if op == "exists":
        matched = resolved is not MISSING
        return PredicateOutcome(
            matched, f"`{path}` exists -> {matched} ({_describe(resolved)})"
        )

    if "value" not in predicate:
        return PredicateOutcome(False, f"op `{op}` requires `value`")
    expected = predicate["value"]

    if resolved is MISSING:
        # An absent key fails every value-taking operator. "absent OR equals
        # X" is two predicates, written as two, rather than a special case
        # here that every rule then inherits.
        return PredicateOutcome(
            False, f"`{path}` {op} `{expected!r}` -> False (state: absent)"
        )

    if op == "eq":
        matched = resolved == expected
    elif op == "neq":
        matched = resolved != expected
    else:  # "in"
        if not isinstance(expected, list | tuple | set):
            # A string here would substring-match, so "region in 'US'" would
            # quietly admit "U". Refused rather than narrowed.
            return PredicateOutcome(
                False,
                f"`in` requires list/tuple/set value, got {type(expected).__name__}",
            )
        try:
            matched = resolved in expected
        except TypeError:  # unhashable or unorderable -> closed
            matched = False

    return PredicateOutcome(
        matched,
        f"`{path}` {op} `{expected!r}` -> {matched} "
        f"(state: {'matches' if matched else 'present, differs'})",
    )

skill_applies 

skill_applies(state: Any, precondition: Any) -> bool

Whether every state predicate on a skill matches state.

A skill with no state predicates applies everywhere -- that is what writing none means, and it keeps the bare-identifier form untouched by this module.

Source code in src/symfonic/memory/layers/procedural/predicates.py 
214
215
216
217
218
219
220
221
222
223
224
def skill_applies(state: Any, precondition: Any) -> bool:
    """Whether every state predicate on a skill matches ``state``.

    A skill with no state predicates applies everywhere -- that is what
    writing none means, and it keeps the bare-identifier form untouched by
    this module.
    """
    return all(
        evaluate_state_predicate(state, predicate).matched
        for predicate in state_predicates_in(precondition)
    )

state_predicates_in 

state_predicates_in(precondition: Any) -> tuple[dict[str, Any], ...]

The state predicates in a precondition list, unwrapped from state.

The other half of the list -- bare tool identifiers -- is a different question with a different answer (has this tool been called yet?), so the two are partitioned rather than evaluated together.

Source code in src/symfonic/memory/layers/procedural/predicates.py 
198
199
200
201
202
203
204
205
206
207
208
209
210
211
def state_predicates_in(precondition: Any) -> tuple[dict[str, Any], ...]:
    """The state predicates in a precondition list, unwrapped from ``state``.

    The other half of the list -- bare tool identifiers -- is a different
    question with a different answer (has this tool been called yet?), so the
    two are partitioned rather than evaluated together.
    """
    if not isinstance(precondition, list):
        return ()
    return tuple(
        item["state"]
        for item in precondition
        if isinstance(item, dict) and isinstance(item.get("state"), dict)
    )