symfonic.memory.layers.procedural

procedural

Procedural memory layer -- skills, workflows, and capability routing.

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 symfonic/memory/layers/procedural/router.py

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

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

route async

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

Route a query to appropriate tools.

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

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
query	str	The user's query or request.	required
conversation_history	list[Any] | None	Optional recent conversation for context.	None
callback_manager	CallbackManager | None	v7.4.3 (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 symfonic/memory/layers/procedural/router.py

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

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

    Args:
        scope: Tenant scope for isolation.
        query: The user's query or request.
        conversation_history: Optional recent conversation for context.
        callback_manager: v7.4.3 (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")

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 symfonic/memory/layers/procedural/layer.py

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

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

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

approve_skill async

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

Promote a draft skill to approved+active.

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

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

delete async

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

Delete a skill node from the graph.

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

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

link async

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

No-op. Skills are standalone graph nodes.

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

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

query_skills async

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

Query procedural skills by word-level matching.

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

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

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

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

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

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

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

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

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

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

reject_skill async

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

Mark a draft skill as rejected+inactive.

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

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

render_for_prompt

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

Render a procedural entry for MEMORY_CONTEXT.

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

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

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

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

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

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

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

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

retrieve async

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

Retrieve skills matching the query.

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

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

seed_authored_skill async

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

Seed a hand-authored procedural skill (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 symfonic/memory/layers/procedural/layer.py

async def seed_authored_skill(
    self,
    scope: TenantScope,
    skill: MemoryEntry,
    *,
    source_tag: str,
    precondition: str | list[str | dict[str, Any]] | None = None,
) -> MemoryNode:
    """Seed a hand-authored procedural skill (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 symfonic/memory/layers/procedural/layer.py

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

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

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

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

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

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

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

summarize async

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

Not meaningful for procedural skills. Returns empty string.

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

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

write async

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

Write a skill entry to the graph store.

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

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

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 symfonic/memory/layers/procedural/catalog.py

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

get

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

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

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

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

list_all

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

Return all registered tools.

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

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

register

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

Register a tool with its description and schema.

Parameters:

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

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

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

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

search

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

Search tools by keyword matching against ID and description.

Simple case-insensitive substring matching for v1.

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

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

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