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

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

        observability_hook = NoOpObservabilityHook()
    self._observability_hook = observability_hook

route async

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

Route a query to appropriate tools.

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

Parameters:

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

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 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
    # Store/dedup and review must share one in-process critical section.
    # A multi-process composition still needs a graph CAS port.
    from symfonic.memory.layers.procedural.review_lock import review_lock_for
    self._review_lock = review_lock_for(graph_store.backend)

amend_skill async

amend_skill(scope: TenantScope, node_id: NodeId, *, steps: Any = None, properties: dict[str, Any] | None = None, active: bool | None = None, toggle_active: bool = False) -> MemoryNode

Apply a UI edit through the lifecycle lock without bypassing review.

HTTP callers can edit content metadata and disable an approved skill, but cannot mint review authority or promote a draft by writing graph fields directly. The latter remains the explicit review transition.

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

async def amend_skill(
    self,
    scope: TenantScope,
    node_id: NodeId,
    *,
    steps: Any = None,
    properties: dict[str, Any] | None = None,
    active: bool | None = None,
    toggle_active: bool = False,
) -> MemoryNode:
    """Apply a UI edit through the lifecycle lock without bypassing review.

    HTTP callers can edit content metadata and disable an approved skill,
    but cannot mint review authority or promote a draft by writing graph
    fields directly.  The latter remains the explicit review transition.
    """
    from symfonic.memory.layers.procedural.lifecycle import procedure_lifecycle
    reserved = {
        "status", "active", "procedure_review_authority", "procedure_review_actor",
        "procedure_lifecycle_origin",
    }
    async with self._review_lock:
        node = await self._graph.get_node(scope, node_id)
        if node is None:
            raise KeyError(f"Procedural node {node_id} not found")
        merged = dict(node.properties or {})
        if steps is not None:
            merged["steps"] = steps
        if properties is not None:
            nested = properties.get("record_metadata")
            if isinstance(nested, dict):
                nested = {
                    key: value for key, value in nested.items()
                    if key not in reserved
                }
            merged.update(
                {
                    key: (nested if key == "record_metadata" else value)
                    for key, value in properties.items() if key not in reserved
                }
            )
        # A nested producer payload cannot amend lifecycle indirectly.
        nested_existing = merged.get("record_metadata")
        if isinstance(nested_existing, dict):
            merged["record_metadata"] = {
                key: value for key, value in nested_existing.items()
                if key not in reserved
            }
        status, current_active = procedure_lifecycle(merged)
        if toggle_active:
            # A draft/rejection cannot become active by a router toggle.
            merged["active"] = False if status != "approved" else not current_active
        elif active is not None:
            # Disabling/re-enabling a previously approved procedure does
            # not constitute approval; a draft/rejection still fails closed.
            merged["active"] = active if status == "approved" else False
        merged["_last_edited_by"] = "user_manual_edit"
        return await self._graph.update_node(scope, node_id, {"properties": merged})

approve_skill async

approve_skill(scope: TenantScope, node_id: NodeId, *, action_tool: str | None = None, precondition: Any = None) -> MemoryNode

Apply a human approval and any binding in one locked transition.

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

async def approve_skill(
    self,
    scope: TenantScope,
    node_id: NodeId,
    *,
    action_tool: str | None = None,
    precondition: Any = None,
) -> MemoryNode:
    """Apply a human approval and any binding in one locked transition."""
    async with self._review_lock:
        return await self._set_reviewed_status(
            scope,
            node_id,
            "approved",
            action_tool=action_tool,
            precondition=precondition,
        )

decide_draft_skill async

decide_draft_skill(scope: TenantScope, node_id: NodeId, *, status: str, action_tool: str | None = None, precondition: Any = None, expected_source: object | None = None, expected_source_present: bool = True, expected_action_tool: str | None = None, expected_action_tool_present: bool = False, actor: str | None = None) -> MemoryNode | None

Apply an opt-in automated decision only if the row remains a draft.

This is intentionally limited to one process: its value is that a policy and the human admin door composed over this same layer cannot race a stale decision past a review. Multi-process automation needs a backend compare-and-set port before it can make that claim.

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

async def decide_draft_skill(
    self,
    scope: TenantScope,
    node_id: NodeId,
    *,
    status: str,
    action_tool: str | None = None,
    precondition: Any = None,
    expected_source: object | None = None,
    expected_source_present: bool = True,
    expected_action_tool: str | None = None,
    expected_action_tool_present: bool = False,
    actor: str | None = None,
) -> MemoryNode | None:
    """Apply an opt-in automated decision only if the row remains a draft.

    This is intentionally limited to one process: its value is that a
    policy and the human admin door composed over this same layer cannot
    race a stale decision past a review. Multi-process automation needs a
    backend compare-and-set port before it can make that claim.
    """
    from symfonic.memory.layers.procedural.lifecycle import procedure_lifecycle
    _validate_draft_decision(status, action_tool, precondition)
    if not isinstance(actor, str) or not actor.strip():
        raise ValueError("automatic procedure compare-and-set requires an actor")
    async with self._review_lock:
        node = await self._graph.get_node(scope, node_id)
        if node is None:
            raise KeyError(f"Procedural node {node_id} not found")
        if procedure_lifecycle(node.properties)[0] != "draft":
            return None
        # A policy evaluated an earlier snapshot. It may act only if the
        # fields it was allowed to inspect are unchanged in this locked,
        # authoritative read. ``source`` is provenance, not authority.
        if (
            ("source" in node.properties) != expected_source_present
            or (
                expected_source_present
                and node.properties.get("source") != expected_source
            )
            or ("action_tool" in node.properties) != expected_action_tool_present
            or (
                expected_action_tool_present
                and node.properties.get("action_tool") != expected_action_tool
            )
        ):
            return None
        merged = dict(node.properties or {})
        if action_tool is not None:
            merged["action_tool"] = action_tool
        if precondition is not None:
            merged["precondition"] = precondition
        merged["status"] = status
        merged["active"] = status == "approved"
        merged["procedure_review_authority"] = "policy"
        merged["procedure_review_actor"] = actor
        return await self._graph.update_node(scope, node_id, {"properties": merged})

deduplicate_skills async

deduplicate_skills(scope: TenantScope, *, audit: Any = None, context: Any = None) -> tuple[int, list[MemoryNode]]

Refuse an unattributed pass before the lock; the janitor gates and audits.

The atomic gate and the one awaited intent live at the janitor's shared boundary, so this layer neither re-gates nor records its own intent.

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

async def deduplicate_skills(
    self, scope: TenantScope, *, audit: Any = None, context: Any = None,
) -> tuple[int, list[MemoryNode]]:
    """Refuse an unattributed pass before the lock; the janitor gates and audits.

    The atomic gate and the one awaited intent live at the janitor's shared
    boundary, so this layer neither re-gates nor records its own intent.
    """
    from symfonic.memory.layers.procedural.dedup_audit import require_dedup_audit
    require_dedup_audit(audit, context)
    async with self._review_lock:
        return await self._janitor.deduplicate_all(
            scope, self._graph, audit=audit, context=context,
        )

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.delete_skill(scope, node_id)

delete_skill async

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

Delete through the same writer gate as review and deduplication.

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

async def delete_skill(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete through the same writer gate as review and deduplication."""
    async with self._review_lock:
        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."""
    async with self._review_lock:
        return await self._set_reviewed_status(scope, node_id, "rejected")

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)

retrieve_report async

retrieve_report(scope: TenantScope, query: str, top_k: int = 5, *, include_drafts: bool = False)

Opt-in per-row origin and queried scope; no routing policy change.

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

async def retrieve_report(
    self, scope: TenantScope, query: str, top_k: int = 5,
    *, include_drafts: bool = False,
):
    """Opt-in per-row origin and queried scope; no routing policy change."""
    from symfonic.memory.layers.procedural.retrieval import query_report

    return await query_report(self, scope, query, top_k, include_drafts=include_drafts)

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

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_legacy_consolidated_skill async

store_legacy_consolidated_skill(scope: TenantScope, skill: MemoryEntry, *, status: str) -> MemoryNode

Persist one retired-engine terminal with explicit compatibility intent.

The compatibility writer is the sole minting path for the automatic provenance marker. Generic writes (including HTTP creation) retain their supplied lifecycle but remain unsealed historical terminals.

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

async def store_legacy_consolidated_skill(
    self, scope: TenantScope, skill: MemoryEntry, *, status: str,
) -> MemoryNode:
    """Persist one retired-engine terminal with explicit compatibility intent.

    The compatibility writer is the sole minting path for the automatic
    provenance marker.  Generic writes (including HTTP creation) retain
    their supplied lifecycle but remain unsealed historical terminals.
    """
    if status not in {"approved", "rejected"}:
        raise ValueError("legacy consolidation writes require a terminal status")
    return await self.store_skill(
        scope, skill, status=status, _lifecycle_origin="legacy_consolidation"
    )

store_phase_12_skill async

store_phase_12_skill(scope: TenantScope, skill: MemoryEntry) -> MemoryNode

Store a replayable Phase-12 draft without trusting entry metadata.

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

async def store_phase_12_skill(
    self, scope: TenantScope, skill: MemoryEntry,
) -> MemoryNode:
    """Store a replayable Phase-12 draft without trusting entry metadata."""
    return await self.store_skill(
        scope, skill, status="draft", _lifecycle_origin="phase_12_automatic"
    )

store_skill async

store_skill(scope: TenantScope, skill: MemoryEntry, *, status: str = 'draft', _lifecycle_origin: str | None = None, _return_created: bool = False) -> MemoryNode | tuple[MemoryNode, bool]

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'
_return_created	bool	Internal native-create option; report the insert-versus-merge decision made under the writer lock.	False

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

async def store_skill(
    self, scope: TenantScope, skill: MemoryEntry,
    *, status: str = "draft", _lifecycle_origin: str | None = None,
    _return_created: bool = False,
) -> MemoryNode | tuple[MemoryNode, bool]:
    """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).
        _return_created: Internal native-create option; report the
                insert-versus-merge decision made under the writer lock.
    """
    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",
                    # Only the review-domain transition can mint this seal.
                    "procedure_review_authority",
                    # Lifecycle provenance is domain-owned, never a
                    # producer-controlled metadata claim.
                    "procedure_lifecycle_origin",
                )
            },
            # Caller ownership overrides producer claims before the janitor.
            "scope_path": scope.scope_path,
        },
        importance=skill.importance,
    )
    if _lifecycle_origin == "legacy_consolidation" and status in {"approved", "rejected"}:
        # This is deliberately narrower than ``store_skill(status=...)``.
        # HTTP/import callers can create a terminal row, but cannot claim
        # it came from the retired engine's automatic consolidation path.
        node.properties["procedure_lifecycle_origin"] = "legacy_auto_approved"
    elif _lifecycle_origin == "phase_12_automatic" and status == "draft":
        # Only the Phase-12 domain helper below may mark its replayable
        # draft as automatic.  ``source`` remains descriptive provenance,
        # not lifecycle authority.
        node.properties["procedure_lifecycle_origin"] = _lifecycle_origin
    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,
    )

    async with self._review_lock:
        sealed = await self._sealed_procedure_for_supplied_id(scope, skill.node_id)
        if sealed is not None:
            # An explicit identifier is an upsert key in every supported
            # backend.  Do not let a producer, HTTP request, or retired
            # consolidation replay replace a terminal decision minted by
            # the review domain merely by reusing that key.
            from symfonic.memory.layers.procedural.review_lock import durable_backend_for
            visible = await durable_backend_for(self._graph.backend).get_node(scope, sealed.id)
            if visible is None:
                raise ValueError("procedure identifier is unavailable in this scope")
            return (sealed, False) if _return_created else sealed
        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, False) if _return_created else canonical
        inserted = await self._graph.add_node(scope, node)
        return (inserted, True) if _return_created else inserted

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)

ToolCatalog

ToolCatalog()

Registry of available tools and capabilities.

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

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

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

get

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

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

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

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

list_all

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

Return all registered tools.

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

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

register

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

Register a tool with its description and schema.

Parameters:

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

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

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

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

search

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

Search tools by keyword matching against ID and description.

Simple case-insensitive substring matching for v1.

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

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

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