symfonic.memory.layers.procedural.layer

layer

ProceduralLayer -- skills, workflows, and learned procedures.

Stores skills as graph nodes with steps and context properties. Implements BaseMemoryStore with summarize and link as no-ops.

ProceduralLayer

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

Procedural memory layer for skills and workflows.

BaseMemoryStore contract

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

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

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

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

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

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

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

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

approve_skill async

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

Promote a draft skill to approved+active.

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

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

delete async

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

Delete a skill node from the graph.

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

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

link async

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

No-op. Skills are standalone graph nodes.

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

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

query_skills async

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

Query procedural skills by word-level matching.

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

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

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

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

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

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

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

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

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

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

reject_skill async

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

Mark a draft skill as rejected+inactive.

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

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

render_for_prompt

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

Render a procedural entry for MEMORY_CONTEXT.

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

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

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

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

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

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

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

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

retrieve async

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

Retrieve skills matching the query.

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

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

seed_authored_skill async

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

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

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

Parameters:

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

Returns:

Type	Description
MemoryNode	The persisted :class:MemoryNode.

Raises:

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

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

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

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

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

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

    Returns:
        The persisted :class:`MemoryNode`.

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

store_skill async

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

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

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

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

Parameters:

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

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

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

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

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

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

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

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

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

summarize async

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

Not meaningful for procedural skills. Returns empty string.

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

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

write async

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

Write a skill entry to the graph store.

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

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