symfonic.memory.layers

layers

Memory layer implementations (Pentad model).

Re-exports all layer classes for convenient importing.

EpisodicLayer

EpisodicLayer(
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    telemetry_sink: EpisodicTelemetrySink | None = None,
)

Episodic memory layer for narrative events and history.

BaseMemoryStore contract

• retrieve: queries events by vector similarity
• write: stores an event with its embedding
• summarize: returns empty string (delegates to compaction engine)
• link: no-op (episodes are vector-stored, not graph-linked)
• delete: removes an event from the vector store

Source code in src/symfonic/memory/layers/episodic.py

def __init__(
    self,
    vector_backend: VectorBackend,
    embedding_provider: EmbeddingProvider,
    telemetry_sink: EpisodicTelemetrySink | None = None,
) -> None:
    self._vector = vector_backend
    self._embedder = embedding_provider
    # When None (the default), the write path skips record construction
    # entirely -- zero cost on the hot path, which is the v6.1 contract.
    self._telemetry_sink = telemetry_sink
    # v7.22 T-7.22.8 -- emit-once-per-(tenant, sub_tenant) tracking for
    # MultiScopeIgnoredWarning.  See T-7.22.11 for the non-atomic
    # check-then-add concurrency rationale.
    self._multi_scope_warned: set[tuple[str, str | None]] = set()

count_events async

count_events(scope: TenantScope) -> int

Return the total number of stored episodic events for the tenant.

Delegates to VectorBackend.count() when available, otherwise falls back to 0 to avoid a full scan with an arbitrary embedding.

Source code in src/symfonic/memory/layers/episodic.py

async def count_events(self, scope: TenantScope) -> int:
    """Return the total number of stored episodic events for the tenant.

    Delegates to VectorBackend.count() when available, otherwise falls back
    to 0 to avoid a full scan with an arbitrary embedding.
    """
    if hasattr(self._vector, "count"):
        return await self._vector.count(scope)
    return 0

delete async

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

Delete an episodic event by its ID.

Source code in src/symfonic/memory/layers/episodic.py

async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete an episodic event by its ID."""
    await self._vector.delete(scope, [str(node_id)])

link async

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

No-op. Episodes are vector-stored, not graph-linked.

Source code in src/symfonic/memory/layers/episodic.py

async def link(
    self, scope: TenantScope, source: NodeId, target: NodeId, relationship: str
) -> None:
    """No-op. Episodes are vector-stored, not graph-linked."""
    return

list_events async

list_events(
    scope: TenantScope, limit: int = 1000
) -> list[MemoryEntry]

Return up to limit episodic events for the tenant.

Uses a neutral zero-vector query so that all stored events are returned in approximate descending-similarity order (deterministic for vectors with the same embedding, arbitrary but consistent otherwise). Prefer this over query_events when you need a full listing rather than a relevance-ranked subset.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
limit	int	Maximum number of events to return (default 1000).	1000

Returns:

Type	Description
list[MemoryEntry]	List of MemoryEntry instances (order: most-similar first).

Notes

v7.13.6: when the embedder rejects embed("") (the OpenAI embeddings endpoint returns 400 "input cannot be an empty string") we transparently fall back to embed(" "). The single-space probe is non-empty for OpenAI's validator while remaining distance-agnostic enough to enumerate stored events via similarity ranking. Sentence-transformers and other backends that accept empty strings continue to use the zero-string path. Fixes the v7.13.4 quick_nap phase-10 WARN reported by Jarvio.

Source code in src/symfonic/memory/layers/episodic.py

async def list_events(
    self, scope: TenantScope, limit: int = 1000
) -> list[MemoryEntry]:
    """Return up to *limit* episodic events for the tenant.

    Uses a neutral zero-vector query so that all stored events are
    returned in approximate descending-similarity order (deterministic
    for vectors with the same embedding, arbitrary but consistent
    otherwise).  Prefer this over ``query_events`` when you need a
    full listing rather than a relevance-ranked subset.

    Args:
        scope: Tenant scope for isolation.
        limit: Maximum number of events to return (default 1000).

    Returns:
        List of MemoryEntry instances (order: most-similar first).

    Notes:
        v7.13.6: when the embedder rejects ``embed("")`` (the OpenAI
        embeddings endpoint returns 400 ``"input cannot be an empty
        string"``) we transparently fall back to ``embed(" ")``.  The
        single-space probe is non-empty for OpenAI's validator while
        remaining distance-agnostic enough to enumerate stored events
        via similarity ranking.  Sentence-transformers and other
        backends that accept empty strings continue to use the
        zero-string path.  Fixes the v7.13.4 quick_nap phase-10 WARN
        reported by Jarvio.
    """
    count = await self.count_events(scope)
    if count == 0:
        return []
    top_k = min(count, limit)
    # Use a zero-vector query so results are distance-agnostic on
    # identical embeddings (stub backends) and approximately uniform
    # across real backends.  Cosine similarity of a zero vector is
    # undefined (0/0); real backends handle this gracefully by
    # returning 0.0 similarity, which still enumerates all documents.
    # v7.13.6: OpenAI embeddings rejects ``""`` with HTTP 400; retry
    # once with a single-space probe so list_events is safe against
    # any backend regardless of empty-input policy.
    try:
        zero_vec = await self._embedder.embed("")
    except Exception:
        zero_vec = await self._embedder.embed(" ")
    results = await self._vector.search(scope, zero_vec, top_k)
    entries: list[MemoryEntry] = []
    for result in results:
        entries.append(
            MemoryEntry(
                id=result["id"],
                layer=MemoryLayer.EPISODIC,
                tenant_id=scope.tenant_id,
                content=result.get("document", ""),
                score=result.get("score"),
                metadata=result.get("metadata", {}),
            )
        )
    return entries

query_events async

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

Query episodic events by vector similarity.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
query	str	Natural language query to search against. Must be a non-empty, non-whitespace string.	required
top_k	int	Maximum number of results to return.	5

Returns:

Type	Description
list[MemoryEntry]	List of MemoryEntry instances ranked by similarity.

Raises:

Type	Description
ValueError	If query is empty or whitespace-only, or if the embedder returns an empty vector.

Source code in src/symfonic/memory/layers/episodic.py

async def query_events(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Query episodic events by vector similarity.

    Args:
        scope: Tenant scope for isolation.
        query: Natural language query to search against.  Must be a
            non-empty, non-whitespace string.
        top_k: Maximum number of results to return.

    Returns:
        List of MemoryEntry instances ranked by similarity.

    Raises:
        ValueError: If ``query`` is empty or whitespace-only, or if the
            embedder returns an empty vector.
    """
    if not query.strip():
        raise ValueError("query_events requires a non-empty query string")
    query_embedding = await self._embedder.embed(query)
    if not query_embedding:
        raise ValueError("embedder returned an empty vector for query")
    results = await self._vector.search(scope, query_embedding, top_k)
    entries: list[MemoryEntry] = []
    for result in results:
        entries.append(
            MemoryEntry(
                id=result["id"],
                layer=MemoryLayer.EPISODIC,
                tenant_id=scope.tenant_id,
                content=result.get("document", ""),
                score=result.get("score"),
                metadata=result.get("metadata", {}),
            )
        )
    return entries

render_for_prompt

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

Render an episodic entry for MEMORY_CONTEXT.

v7.7: when metadata['speaker'] is present (per-message row introduced in Phase B), emit [episodic] <speaker>: <content> so the LLM sees a clean speaker-prefixed line. Pre-v7.7 rows (no speaker) or rows tagged speaker == 'legacy' fall back to :func:render_legacy for byte-identical pre-v7.7 output and prompt-cache parity.

Source code in src/symfonic/memory/layers/episodic.py

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

    v7.7: when ``metadata['speaker']`` is present (per-message row
    introduced in Phase B), emit ``[episodic] <speaker>: <content>``
    so the LLM sees a clean speaker-prefixed line.  Pre-v7.7 rows
    (no speaker) or rows tagged ``speaker == 'legacy'`` fall back
    to :func:`render_legacy` for byte-identical pre-v7.7 output
    and prompt-cache parity.
    """
    metadata = getattr(entry, "metadata", {}) or {}
    speaker = metadata.get("speaker")
    if isinstance(speaker, str) and speaker and speaker != "legacy":
        content = str(getattr(entry, "content", ""))
        layer_name = getattr(entry, "layer", "episodic")
        return f"[{layer_name}] {speaker}: {content}"
    return render_legacy(entry, scrubber)

retrieve async

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

Retrieve episodic events matching the query.

v7.22 T-7.22.8: emits :class:MultiScopeIgnoredWarning on first encounter of a scope with inherits_from. Episodic events are speaker-attributed timestamped records bound to a single (tenant, sub_tenant) pair; merging inherited scopes would corrupt per-tenant cost accounting and break GDPR scope guarantees (v5.5.0 Gate 3).

Source code in src/symfonic/memory/layers/episodic.py

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

    v7.22 T-7.22.8: emits :class:`MultiScopeIgnoredWarning` on first
    encounter of a scope with ``inherits_from``.  Episodic events are
    speaker-attributed timestamped records bound to a single
    ``(tenant, sub_tenant)`` pair; merging inherited scopes would
    corrupt per-tenant cost accounting and break GDPR scope
    guarantees (v5.5.0 Gate 3).
    """
    from symfonic.memory._multi_scope_warn import maybe_warn_multi_scope

    maybe_warn_multi_scope(self, self._multi_scope_warned, scope)
    return await self.query_events(scope, query, top_k)

store_event async

store_event(
    scope: TenantScope,
    event: MemoryEntry,
    *,
    source_phase: str = "episodic.store_event",
) -> None

Store an event with its embedding in the vector backend.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
event	MemoryEntry	The memory entry representing the event.	required
source_phase	str	Tag identifying which caller triggered the write (e.g. "engine.consolidate", "phase10"). Recorded on the telemetry record when a sink is configured; ignored otherwise. Default identifies a direct call to this method.	'episodic.store_event'

v6.2 T05 telemetry: when the caller populates event.metadata with any of tokens_in / tokens_out / input_tokens / output_tokens / cost_usd / model / embedding_latency_ms, the values are lifted onto the :class:EpisodicTelemetryRecord for the corpus. Keys missing from metadata remain None and are stripped from the JSONL line (v6.1 shape preserved byte-for-byte). The engine can source the token / cost fields from LLMPreCallEvent / LLMEndEvent and attach them to the metadata before calling store_event; direct callers that do not consume LangChain callbacks may pass the values in metadata themselves.

Source code in src/symfonic/memory/layers/episodic.py

async def store_event(
    self,
    scope: TenantScope,
    event: MemoryEntry,
    *,
    source_phase: str = "episodic.store_event",
) -> None:
    """Store an event with its embedding in the vector backend.

    Args:
        scope: Tenant scope for isolation.
        event: The memory entry representing the event.
        source_phase: Tag identifying which caller triggered the
            write (e.g. ``"engine.consolidate"``, ``"phase10"``).
            Recorded on the telemetry record when a sink is
            configured; ignored otherwise. Default identifies a
            direct call to this method.

    v6.2 T05 telemetry: when the caller populates ``event.metadata``
    with any of ``tokens_in`` / ``tokens_out`` / ``input_tokens`` /
    ``output_tokens`` / ``cost_usd`` / ``model`` /
    ``embedding_latency_ms``, the values are lifted onto the
    :class:`EpisodicTelemetryRecord` for the corpus. Keys missing
    from metadata remain ``None`` and are stripped from the JSONL
    line (v6.1 shape preserved byte-for-byte). The engine can source
    the token / cost fields from ``LLMPreCallEvent`` /
    ``LLMEndEvent`` and attach them to the metadata before calling
    store_event; direct callers that do not consume LangChain
    callbacks may pass the values in ``metadata`` themselves.
    """
    embedding = await self._embedder.embed(event.content)
    doc_id = event.id or str(uuid.uuid4())
    await self._vector.add(
        scope,
        ids=[doc_id],
        embeddings=[embedding],
        metadatas=[{
            "layer": MemoryLayer.EPISODIC.value,
            "importance": event.importance,
            "created_at": event.created_at.isoformat(),
            **event.metadata,
        }],
        documents=[event.content],
    )

    # Telemetry emit -- gated on sink presence so the default path
    # does zero work.  A raising sink must never break the write.
    sink = self._telemetry_sink
    if sink is None:
        return
    try:
        md = event.metadata or {}
        action_raw = md.get("action_type")
        action_type = str(action_raw) if action_raw is not None else None

        # v6.2 T05: lift optional token / cost / model fields out
        # of ``event.metadata`` into first-class record fields.
        # Metadata keys are advisory -- callers that do not source
        # them yield a record with the v6.1 shape.
        def _as_int(v: object) -> int | None:
            try:
                return int(v) if v is not None else None
            except (TypeError, ValueError):
                return None

        def _as_float(v: object) -> float | None:
            try:
                return float(v) if v is not None else None
            except (TypeError, ValueError):
                return None

        # Pydantic TenantScope always has `namespace` as a field
        # (defaulting to None); getattr's fallback never fires. Coalesce
        # explicitly so the corpus always has a non-null partitioning
        # key, which the v6.2 Phase 12 extractor groups on.
        record = build_record(
            tenant_id=scope.tenant_id,
            scope=scope.namespace or "default",
            episode_id=doc_id,
            node_count=1,
            source_phase=source_phase,
            action_type=action_type,
            tokens_in=_as_int(md.get("tokens_in")),
            tokens_out=_as_int(md.get("tokens_out")),
            input_tokens=_as_int(md.get("input_tokens")),
            output_tokens=_as_int(md.get("output_tokens")),
            cost_usd=_as_float(md.get("cost_usd")),
            model=(
                str(md["model"])
                if md.get("model") is not None
                else None
            ),
            embedding_latency_ms=_as_float(md.get("embedding_latency_ms")),
        )
        await sink.emit(record)
    except Exception:
        _telemetry_logger.debug(
            "episodic telemetry emit failed for %s", doc_id, exc_info=True,
        )

summarize async

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

Not directly implemented. Delegates to compaction engine.

Source code in src/symfonic/memory/layers/episodic.py

async def summarize(self, scope: TenantScope, entries: list[MemoryEntry]) -> str:
    """Not directly implemented. Delegates to compaction engine."""
    return ""

write async

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

Write an episodic event to the vector store.

Source code in src/symfonic/memory/layers/episodic.py

async def write(self, scope: TenantScope, entry: MemoryEntry) -> None:
    """Write an episodic event to the vector store."""
    await self.store_event(scope, entry)

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)

ProspectiveLayer

ProspectiveLayer(
    graph_store: GraphMemoryStore,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
)

Prospective memory layer for triggers and reminders.

BaseMemoryStore contract

• retrieve: returns [] -- triggers are NOT generic hydration hits
• write: stores a trigger as a graph node
• summarize: returns empty string
• link: no-op
• delete: removes a trigger node

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

Default None preserves byte-identical pre-Item-13.1 behaviour. When supplied, every set_trigger write runs through :func:maybe_embed so prospective triggers auto-embed when the caller leaves node.embedding=None. Caller-provided embeddings always win.

Per-kind embedding-text rule: triggers always fall into the default branch of :func:symfonic.memory.embeddings.auto_embed.resolve_embedding_text and embed properties['content'] (the trigger's human-readable description) rather than the label.

Source code in src/symfonic/memory/layers/prospective.py

def __init__(
    self,
    graph_store: GraphMemoryStore,
    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 ``set_trigger`` write runs through
    :func:`maybe_embed` so prospective triggers auto-embed when the
    caller leaves ``node.embedding=None``. Caller-provided embeddings
    always win.

    Per-kind embedding-text rule: triggers always fall into the
    default branch of
    :func:`symfonic.memory.embeddings.auto_embed.resolve_embedding_text`
    and embed ``properties['content']`` (the trigger's human-readable
    description) rather than the label.
    """
    self._graph = graph_store
    self._embedding_provider = embedding_provider
    self._embedding_cache = embedding_cache
    # v7.22 T-7.22.10 -- emit-once tracking for MultiScopeIgnoredWarning.
    # Prospective.retrieve() returns [] by design (since v6.x); the
    # warning is no-op-but-consistent so adopters routing a multi-scope
    # scope through every layer see exactly ONE warning per non-
    # procedural layer per tenant -- not three with one silent gap.
    self._multi_scope_warned: set[tuple[str, str | None]] = set()

delete async

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

Delete a trigger node.

Source code in src/symfonic/memory/layers/prospective.py

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

get_pending async

get_pending(scope: TenantScope) -> list[MemoryEntry]

Get all unresolved triggers for the tenant.

Source code in src/symfonic/memory/layers/prospective.py

async def get_pending(self, scope: TenantScope) -> list[MemoryEntry]:
    """Get all unresolved triggers for the tenant."""
    nodes = await self._graph.query_nodes(scope, layer=MemoryLayer.PROSPECTIVE)
    pending: list[MemoryEntry] = []
    for node in nodes:
        if not node.properties.get("resolved", False):
            pending.append(
                MemoryEntry(
                    layer=MemoryLayer.PROSPECTIVE,
                    tenant_id=node.tenant_id,
                    content=str(node.properties.get("content", node.label)),
                    node_id=node.id,
                    importance=node.importance,
                    metadata={
                        "condition": node.properties.get("condition", ""),
                        "task": node.properties.get("task", ""),
                    },
                    created_at=node.created_at,
                )
            )
    return pending

link async

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

No-op. Triggers are standalone nodes.

Source code in src/symfonic/memory/layers/prospective.py

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

render_for_prompt

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

Render a prospective entry for MEMORY_CONTEXT.

v7.7: explicit override delegating to :func:render_legacy for byte-identical pre-v7.7 output. Prospective triggers don't carry the v7.5+ skill-aware metadata signals so a layer-specific shape would have no upside.

Source code in src/symfonic/memory/layers/prospective.py

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

    v7.7: explicit override delegating to :func:`render_legacy`
    for byte-identical pre-v7.7 output.  Prospective triggers
    don't carry the v7.5+ skill-aware metadata signals so a
    layer-specific shape would have no upside.
    """
    return render_legacy(entry, scrubber)

resolve_trigger async

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

Mark a trigger as resolved.

Source code in src/symfonic/memory/layers/prospective.py

async def resolve_trigger(self, scope: TenantScope, node_id: NodeId) -> None:
    """Mark a trigger as resolved."""
    await self._graph.update_node(scope, node_id, {"resolved": True})

retrieve async

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

Return [] -- prospective triggers are not generic hits.

Previously this bulk-dumped every unresolved trigger via get_pending regardless of query, so a bare "Hello" recited all pending reminders into MEMORY_CONTEXT. Pending triggers surface only through the dedicated PROSPECTIVE_TASKS prompt channel and the explicit :meth:get_pending API; they must never enter per-turn semantic hydration. Callers that need the trigger list must call :meth:get_pending directly.

v7.22 T-7.22.10: emits :class:MultiScopeIgnoredWarning on first encounter of a scope carrying inherits_from. Consistency with the other non-procedural layers -- even though the result is always [], adopters wiring a multi-scope scope through every layer get exactly one warning per layer rather than a silent gap.

Source code in src/symfonic/memory/layers/prospective.py

async def retrieve(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Return ``[]`` -- prospective triggers are not generic hits.

    Previously this bulk-dumped every unresolved trigger via
    ``get_pending`` regardless of ``query``, so a bare "Hello"
    recited all pending reminders into ``MEMORY_CONTEXT``. Pending
    triggers surface only through the dedicated ``PROSPECTIVE_TASKS``
    prompt channel and the explicit :meth:`get_pending` API; they
    must never enter per-turn semantic hydration. Callers that need
    the trigger list must call :meth:`get_pending` directly.

    v7.22 T-7.22.10: emits :class:`MultiScopeIgnoredWarning` on first
    encounter of a scope carrying ``inherits_from``.  Consistency with
    the other non-procedural layers -- even though the result is
    always ``[]``, adopters wiring a multi-scope scope through every
    layer get exactly one warning per layer rather than a silent gap.
    """
    from symfonic.memory._multi_scope_warn import maybe_warn_multi_scope

    maybe_warn_multi_scope(self, self._multi_scope_warned, scope)
    return []

set_trigger async

set_trigger(
    scope: TenantScope, trigger: MemoryEntry
) -> MemoryNode

Store a trigger with condition and task properties.

The trigger's metadata should include 'condition' (str) and 'task' (str) keys.

Source code in src/symfonic/memory/layers/prospective.py

async def set_trigger(
    self, scope: TenantScope, trigger: MemoryEntry
) -> MemoryNode:
    """Store a trigger with condition and task properties.

    The trigger's metadata should include 'condition' (str) and
    'task' (str) keys.
    """
    node = MemoryNode(
        layer=MemoryLayer.PROSPECTIVE,
        tenant_id=scope.tenant_id,
        label=trigger.content[:100],
        properties={
            "content": trigger.content,
            "condition": trigger.metadata.get("condition", ""),
            "task": trigger.metadata.get("task", ""),
            "resolved": False,
        },
        importance=trigger.importance,
    )
    if trigger.node_id:
        node.id = trigger.node_id
    # v7.3 Item 13.1: auto-embed when a provider is configured and the
    # caller didn't pre-populate node.embedding.
    # TODO(v7.4): batch embeddings when called in a tight loop.
    node = await maybe_embed(
        node, self._embedding_provider, self._embedding_cache,
    )
    return await self._graph.add_node(scope, node)

summarize async

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

Not meaningful for prospective triggers. Returns empty string.

Source code in src/symfonic/memory/layers/prospective.py

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

write async

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

Write a trigger entry.

Source code in src/symfonic/memory/layers/prospective.py

async def write(self, scope: TenantScope, entry: MemoryEntry) -> None:
    """Write a trigger entry."""
    await self.set_trigger(scope, entry)

SemanticLayer

SemanticLayer(
    graph_store: GraphMemoryStore,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
)

Semantic memory layer for permanent facts and graph entities.

BaseMemoryStore contract

• retrieve: queries facts by content similarity
• write: stores a fact as a graph node
• summarize: returns empty string (facts are not summarizable in isolation)
• link: fully implemented (graph-native operation)
• delete: removes a fact node from the graph

Wrap graph_store with optional auto-embedding (v7.3 Item 13.1).

embedding_provider (default None) preserves byte-identical pre-Item-13.1 behaviour -- the caller is still responsible for populating node.embedding before write. When supplied, the provider is invoked through :func:maybe_embed before every store_fact / write so nodes the caller leaves with embedding=None auto-embed. Caller-provided embeddings always win.

embedding_cache is the shared sha256-keyed LRU cache the orchestrator hands to every graph-backed layer; passing None is supported but means this layer never benefits from a sibling layer having already embedded the same text.

Per-kind embedding-text rule lives in :func:symfonic.memory.embeddings.auto_embed.resolve_embedding_text -- semantic entity nodes (label prefix Entity:) prefer node.label; all other semantic nodes prefer node.properties['content'].

Source code in src/symfonic/memory/layers/semantic.py

def __init__(
    self,
    graph_store: GraphMemoryStore,
    embedding_provider: EmbeddingProvider | None = None,
    embedding_cache: EmbeddingCache | None = None,
) -> None:
    """Wrap *graph_store* with optional auto-embedding (v7.3 Item 13.1).

    ``embedding_provider`` (default ``None``) preserves byte-identical
    pre-Item-13.1 behaviour -- the caller is still responsible for
    populating ``node.embedding`` before write. When supplied, the
    provider is invoked through :func:`maybe_embed` before every
    ``store_fact`` / write so nodes the caller leaves with
    ``embedding=None`` auto-embed. Caller-provided embeddings always
    win.

    ``embedding_cache`` is the shared sha256-keyed LRU cache the
    orchestrator hands to every graph-backed layer; passing ``None``
    is supported but means this layer never benefits from a sibling
    layer having already embedded the same text.

    Per-kind embedding-text rule lives in
    :func:`symfonic.memory.embeddings.auto_embed.resolve_embedding_text`
    -- semantic entity nodes (label prefix ``Entity:``) prefer
    ``node.label``; all other semantic nodes prefer
    ``node.properties['content']``.
    """
    self._graph = graph_store
    self._embedding_provider = embedding_provider
    self._embedding_cache = embedding_cache
    # v7.22 T-7.22.7 -- ``(tenant_id, sub_tenant_id)`` pairs that have
    # already received a MultiScopeIgnoredWarning from this layer
    # instance.  Non-atomic check-then-add is acceptable here
    # (T-7.22.11); a duplicate warning under race is benign and avoids
    # imposing a lock on the per-query hot path.
    self._multi_scope_warned: set[tuple[str, str | None]] = set()

delete async

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

Delete a semantic fact node and its edges.

Source code in src/symfonic/memory/layers/semantic.py

async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Delete a semantic fact node and its edges."""
    await self._graph.delete_node(scope, node_id)

link async

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

Create a relationship edge between two semantic nodes.

Source code in src/symfonic/memory/layers/semantic.py

async def link(
    self, scope: TenantScope, source: NodeId, target: NodeId, relationship: str
) -> None:
    """Create a relationship edge between two semantic nodes."""
    edge = MemoryEdge(
        source=source,
        target=target,
        relationship=relationship,
        tenant_id=scope.tenant_id,
    )
    await self._graph.add_edge(scope, edge)

query_facts async

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

Query semantic facts by keyword matching.

For production use, this should be combined with embedding-based retrieval via the RetrievalEngine. This method provides basic keyword filtering as a fallback.

On a no-keyword-match it returns ONLY the always-on persona node (the AGENT_IDENTITY allowlist) -- never the full importance-sorted node set. The old "top-k by importance" fallback recited unrelated high-importance facts on irrelevant queries (e.g. a bare "Hello"); the allowlist still serves "who are you?"-style identity turns without that leak.

Source code in src/symfonic/memory/layers/semantic.py

async def query_facts(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Query semantic facts by keyword matching.

    For production use, this should be combined with embedding-based
    retrieval via the RetrievalEngine. This method provides basic
    keyword filtering as a fallback.

    On a no-keyword-match it returns ONLY the always-on persona node
    (the ``AGENT_IDENTITY`` allowlist) -- never the full
    importance-sorted node set. The old "top-k by importance"
    fallback recited unrelated high-importance facts on irrelevant
    queries (e.g. a bare "Hello"); the allowlist still serves
    "who are you?"-style identity turns without that leak.
    """
    nodes = await self._graph.query_nodes(scope, layer=MemoryLayer.SEMANTIC)
    query_words = [w for w in query.lower().split() if len(w) > 2]

    def _to_entry(node: MemoryNode) -> MemoryEntry:
        return MemoryEntry(
            layer=MemoryLayer.SEMANTIC,
            tenant_id=node.tenant_id,
            content=str(node.properties.get("content", node.label)),
            node_id=node.id,
            importance=node.importance,
            created_at=node.created_at,
        )

    # Try keyword matching first.
    if query_words:
        matched = [
            node for node in nodes
            if any(
                word in str(node.properties.get("content", node.label)).lower()
                for word in query_words
            )
        ]
        if matched:
            return [_to_entry(n) for n in matched[:top_k]]

    # No keyword match (or an empty query): surface only the
    # always-on persona node(s) via an explicit allowlist instead of
    # the unbounded importance-sorted fallback.
    persona = [n for n in nodes if _is_persona_node(n)]
    return [_to_entry(n) for n in persona[:top_k]]

render_for_prompt

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

Render a semantic entry for MEMORY_CONTEXT.

v7.7: explicit override delegating to :func:render_legacy so the per-layer contract test catches a missing override on a future subclass. The pre-v7.7 [layer] content (k=v) shape is the correct semantic-fact render; layer-specific shaping would change cache-keyed prompt bytes for every adopter.

Source code in src/symfonic/memory/layers/semantic.py

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

    v7.7: explicit override delegating to :func:`render_legacy` so
    the per-layer contract test catches a missing override on a
    future subclass.  The pre-v7.7 ``[layer] content (k=v)`` shape
    is the correct semantic-fact render; layer-specific shaping
    would change cache-keyed prompt bytes for every adopter.
    """
    return render_legacy(entry, scrubber)

retrieve async

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

Retrieve semantic facts matching the query.

v7.22 T-7.22.7: emits :class:MultiScopeIgnoredWarning on first encounter of a scope carrying inherits_from. The SemanticLayer does NOT honour the field (facts are tenant-bound by entity construction); the warning surfaces the silent-ignore class so adopters know their inheritance setup is not propagating here.

Source code in src/symfonic/memory/layers/semantic.py

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

    v7.22 T-7.22.7: emits :class:`MultiScopeIgnoredWarning` on first
    encounter of a scope carrying ``inherits_from``.  The SemanticLayer
    does NOT honour the field (facts are tenant-bound by entity
    construction); the warning surfaces the silent-ignore class so
    adopters know their inheritance setup is not propagating here.
    """
    from symfonic.memory._multi_scope_warn import maybe_warn_multi_scope

    maybe_warn_multi_scope(self, self._multi_scope_warned, scope)
    return await self.query_facts(scope, query, top_k)

store_fact async

store_fact(
    scope: TenantScope, fact: MemoryEntry
) -> MemoryNode

Store a fact as a MemoryNode in the graph.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope for isolation.	required
fact	MemoryEntry	The memory entry representing the fact.	required

Returns:

Type	Description
MemoryNode	The persisted MemoryNode.

Source code in src/symfonic/memory/layers/semantic.py

async def store_fact(self, scope: TenantScope, fact: MemoryEntry) -> MemoryNode:
    """Store a fact as a MemoryNode in the graph.

    Args:
        scope: Tenant scope for isolation.
        fact: The memory entry representing the fact.

    Returns:
        The persisted MemoryNode.
    """
    # Prefer the LLM-provided label (passed via metadata) over the
    # rich_content string so that labels like "SOUL" are preserved
    # instead of becoming "profession: developer, ...".
    node_label = fact.metadata.get("label") or fact.content[:100]
    node = MemoryNode(
        layer=MemoryLayer.SEMANTIC,
        tenant_id=scope.tenant_id,
        label=str(node_label)[:100],
        properties={"content": fact.content, **fact.metadata},
        importance=fact.importance,
    )
    if fact.node_id:
        node.id = fact.node_id
    # v7.3 Item 13.1: auto-embed when a provider is configured and the
    # caller didn't pre-populate node.embedding. Caller-provided
    # embeddings always win; provider failures degrade to embedding=None
    # (see auto_embed.maybe_embed docstring).
    # TODO(v7.4): batch embeddings when add_node is called in a tight
    # loop (EntityLinker mints 5-10 entity nodes per cycle).
    node = await maybe_embed(
        node, self._embedding_provider, self._embedding_cache,
    )
    return await self._graph.add_node(scope, node)

summarize async

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

Not meaningful for semantic facts. Returns empty string.

Source code in src/symfonic/memory/layers/semantic.py

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

write async

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

Write a memory entry as a semantic fact node.

Source code in src/symfonic/memory/layers/semantic.py

async def write(self, scope: TenantScope, entry: MemoryEntry) -> None:
    """Write a memory entry as a semantic fact node."""
    await self.store_fact(scope, entry)

WorkingLayer

WorkingLayer(
    max_entries: int = 10,
    graph_store: object | None = None,
    *,
    stm_summary_mode: STMSummaryMode = "off",
    working_graph_retention: int | None = None,
)

Working memory layer for active conversation context.

BaseMemoryStore contract

• retrieve: returns current context entries (from in-memory buffer)
• write: pushes an entry (FIFO eviction) and mirrors to graph if available
• summarize: returns empty string
• link: no-op
• delete: removes entry from graph when graph_store is set
• clear: clears in-memory deque AND all graph nodes for the tenant

When graph_store is provided, each write is also persisted as a MemoryNode with layer=WORKING so the metrics endpoint can count across restarts.

Graph retention cap: when working_graph_retention is set (an integer), the graph side is capped at that many nodes per tenant. On each push that exceeds the cap, the oldest nodes are pruned. The in-memory deque is governed separately by max_entries (FIFO eviction). The cap is intentionally opt-in (None = unbounded, the pre-F3 behaviour) so existing deployments that rely on the graph for long-term auditing are unaffected until they explicitly set a retention value.

Initialize the layer.

Parameters:

Name	Type	Description	Default
max_entries	int	FIFO buffer ceiling per tenant.	10
graph_store	object | None	Optional graph backend for dual-write persistence.	None
stm_summary_mode	STMSummaryMode	v7.0.1 R3. Controls the default behaviour of :meth:summarize. "off" (default) returns the empty string, preserving v6.1.x byte-for-byte behaviour for direct-API callers. "extractive" routes through the deterministic first-sentence / last-sentence heuristic. "llm" also routes through the extractive path from this method because the in-layer summarize() is sync- safe and cannot wire an async LLM; callers that want LLM-mode summarisation must call :func:symfonic.memory.working.stm_summary.summarize_entries directly with a model_provider. Prior to v7.0.1 this kwarg was ignored: WorkingLayer.summarize always emitted an extractive summary regardless of config, which silently violated the documented contract for direct-API callers that set stm_summary_mode="off".	'off'

Source code in src/symfonic/memory/layers/working.py

def __init__(
    self,
    max_entries: int = 10,
    graph_store: object | None = None,
    *,
    stm_summary_mode: STMSummaryMode = "off",
    working_graph_retention: int | None = None,
) -> None:
    """Initialize the layer.

    Args:
        max_entries: FIFO buffer ceiling per tenant.
        graph_store: Optional graph backend for dual-write persistence.
        stm_summary_mode: v7.0.1 R3. Controls the default behaviour of
            :meth:`summarize`. ``"off"`` (default) returns the empty
            string, preserving v6.1.x byte-for-byte behaviour for
            direct-API callers. ``"extractive"`` routes through the
            deterministic first-sentence / last-sentence heuristic.
            ``"llm"`` also routes through the extractive path from
            this method because the in-layer summarize() is sync-
            safe and cannot wire an async LLM; callers that want
            LLM-mode summarisation must call
            :func:`symfonic.memory.working.stm_summary.summarize_entries`
            directly with a model_provider.  Prior to v7.0.1 this
            kwarg was ignored: ``WorkingLayer.summarize`` always
            emitted an extractive summary regardless of config,
            which silently violated the documented contract for
            direct-API callers that set ``stm_summary_mode="off"``.
    """
    self._max = max_entries
    self._buffers: dict[str, deque[MemoryEntry]] = defaultdict(
        lambda: deque(maxlen=max_entries)
    )
    self._graph_store = graph_store  # Optional GraphMemoryStore for persistence
    self._stm_summary_mode: STMSummaryMode = stm_summary_mode
    # Graph-side retention cap (None = unbounded).  When set, each push
    # prunes the oldest nodes for that tenant once count exceeds the cap.
    self._graph_retention: int | None = working_graph_retention
    # v7.22 T-7.22.9 -- emit-once tracking for MultiScopeIgnoredWarning.
    # WIRED IN ``retrieve`` ONLY (plan §2.d) -- ``push`` / ``get_context``
    # have no scope-merge semantics so emitting from them would be a
    # false positive in adopter mental models.  Non-atomic check-then-add
    # is acceptable (T-7.22.11).
    self._multi_scope_warned: set[tuple[str, str | None]] = set()

clear async

clear(scope: TenantScope) -> None

Clear all working context for the tenant.

Empties the in-memory deque AND deletes all WORKING-layer graph nodes for this tenant when a graph_store is configured. Nodes belonging to other tenants are untouched (tenant isolation is enforced by the query_nodes filter).

Source code in src/symfonic/memory/layers/working.py

async def clear(self, scope: TenantScope) -> None:
    """Clear all working context for the tenant.

    Empties the in-memory deque AND deletes all WORKING-layer graph nodes
    for this tenant when a graph_store is configured.  Nodes belonging to
    other tenants are untouched (tenant isolation is enforced by the
    query_nodes filter).
    """
    key = self._buffer_key(scope)
    if key in self._buffers:
        self._buffers[key].clear()

    if self._graph_store is not None:
        try:
            nodes = await self._graph_store.query_nodes(  # type: ignore[union-attr]
                scope, layer=MemoryLayer.WORKING,
            )
            for node in nodes:
                try:
                    await self._graph_store.delete_node(scope, node.id)  # type: ignore[union-attr]
                except Exception:
                    logger.warning(
                        "Working graph clear: failed to delete node=%s tenant=%s",
                        node.id, scope.tenant_id, exc_info=True,
                    )
        except Exception:
            logger.warning(
                "Working graph clear failed for tenant=%s",
                scope.tenant_id, exc_info=True,
            )

delete async

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

Remove a working memory node from the graph backend.

The in-memory deque uses FIFO eviction only; this method targets the graph-persistence side so that explicit deletes (e.g. triggered by retention-cap pruning or a user request) actually remove the node from the durable store.

When graph_store is not configured the call is a no-op (backward compatible with deployments that use in-memory-only working memory).

Source code in src/symfonic/memory/layers/working.py

async def delete(self, scope: TenantScope, node_id: NodeId) -> None:
    """Remove a working memory node from the graph backend.

    The in-memory deque uses FIFO eviction only; this method targets
    the graph-persistence side so that explicit deletes (e.g. triggered
    by retention-cap pruning or a user request) actually remove the node
    from the durable store.

    When graph_store is not configured the call is a no-op (backward
    compatible with deployments that use in-memory-only working memory).
    """
    if self._graph_store is None:
        return
    try:
        await self._graph_store.delete_node(scope, node_id)  # type: ignore[union-attr]
    except Exception:
        logger.warning(
            "Working graph delete failed for tenant=%s node=%s",
            scope.tenant_id, node_id, exc_info=True,
        )

get_context async

get_context(scope: TenantScope) -> list[MemoryEntry]

Get all current working context entries for the tenant.

Returns entries in chronological order (oldest first).

Source code in src/symfonic/memory/layers/working.py

async def get_context(self, scope: TenantScope) -> list[MemoryEntry]:
    """Get all current working context entries for the tenant.

    Returns entries in chronological order (oldest first).
    """
    key = self._buffer_key(scope)
    return list(self._buffers.get(key, deque()))

link async

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

No-op. Working memory entries are not graph-linked.

Source code in src/symfonic/memory/layers/working.py

async def link(
    self, scope: TenantScope, source: NodeId, target: NodeId, relationship: str
) -> None:
    """No-op. Working memory entries are not graph-linked."""
    return

push async

push(
    scope: TenantScope,
    entry: MemoryEntry,
    *,
    durability: Durability | None = None,
) -> None

Push an entry to the working memory buffer.

If the buffer is at capacity, the oldest entry is evicted (FIFO). If a graph_store is available, the entry is also persisted as a node. When working_graph_retention is set, the oldest graph nodes for this tenant are pruned after the write so the count stays at or below the configured cap.

v7.26.2 (Shape C1): the optional durability kwarg marks the dual-written graph node so the sleep-consolidation pipeline can skip promotion of transient/expired rows. None (default) writes NO durability key -- byte-identical to pre-v7.26.2 (the node reads as "durable" on the consolidation side). Resolution order: explicit kwarg > entry.metadata['durability'] > unset. An invalid value raises :class:ValueError at the call site (fail-loud, never a silent gate bypass).

Source code in src/symfonic/memory/layers/working.py

async def push(
    self,
    scope: TenantScope,
    entry: MemoryEntry,
    *,
    durability: Durability | None = None,
) -> None:
    """Push an entry to the working memory buffer.

    If the buffer is at capacity, the oldest entry is evicted (FIFO).
    If a graph_store is available, the entry is also persisted as a node.
    When *working_graph_retention* is set, the oldest graph nodes for
    this tenant are pruned after the write so the count stays at or
    below the configured cap.

    v7.26.2 (Shape C1): the optional ``durability`` kwarg marks the
    dual-written graph node so the sleep-consolidation pipeline can
    skip promotion of transient/expired rows.  ``None`` (default)
    writes NO durability key -- byte-identical to pre-v7.26.2 (the
    node reads as ``"durable"`` on the consolidation side).  Resolution
    order: explicit kwarg > ``entry.metadata['durability']`` > unset.
    An invalid value raises :class:`ValueError` at the call site
    (fail-loud, never a silent gate bypass).
    """
    # Resolve durability (fail-loud) BEFORE mutating buffer state so an
    # invalid marker never leaves a half-applied write.
    resolved_durability: Durability | None = None
    if durability is not None:
        resolved_durability = coerce_durability(durability)
    elif entry.metadata:
        md_value = entry.metadata.get("durability")
        if md_value is not None:
            resolved_durability = coerce_durability(md_value)

    key = self._buffer_key(scope)
    self._buffers[key].append(entry)

    if self._graph_store is not None:
        try:
            label = (entry.content[:80] if entry.content else "working-context")
            props: dict[str, object] = {
                "content": entry.content,
                **({k: str(v) for k, v in entry.metadata.items()}
                   if entry.metadata else {}),
            }
            # Only write the key when opted in; preserves byte-identical
            # dict shape for the default path (Contract B).
            if resolved_durability is not None:
                props["durability"] = resolved_durability
            node = MemoryNode(
                layer=MemoryLayer.WORKING,
                tenant_id=scope.tenant_id,
                label=label,
                properties=props,
                importance=max(1.0, min(10.0, entry.importance)),
            )
            await self._graph_store.add_node(scope, node)  # type: ignore[union-attr]
            logger.info(
                "Working node persisted to graph: tenant=%s label=%s",
                scope.tenant_id, label[:50],
            )

            # Prune oldest graph nodes when retention cap is configured.
            if self._graph_retention is not None:
                await self._prune_graph_to_cap(scope)

        except Exception:
            logger.warning(
                "Working graph write failed for tenant=%s",
                scope.tenant_id, exc_info=True,
            )
    else:
        logger.debug(
            "Working layer has no graph_store, skipping persistence"
        )

rehydrate_from_messages async

rehydrate_from_messages(
    scope: TenantScope,
    messages: list[Any],
    *,
    max_turns: int | None = None,
) -> int

Replay the tail of a persisted transcript into the working deque.

v7.27.0 restart-resume (Q7, option-b): on session resume the in-memory deque is empty (process restarted), but the durable checkpointer still holds state['messages']. This method reconstructs the RECENCY WINDOW only -- the last N user/assistant turns -- as MemoryEntry rows tagged with the SAME speaker/turn_id/source metadata shape the live auto-turn writes use (engine.py), so downstream consumers cannot tell a rehydrated row from a freshly-written one.

Scope of restore (LOCKED, plan §4.c): recency window ONLY. Tool / system messages are skipped (they are not conversational turns and would blow the cap with noise -- R4). Durable HMS state is NOT touched (it was never lost). No graph write is made -- this is a pure in-memory deque replay; persisting these rows again would double-count graph nodes.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Tenant scope selecting the target deque.	required
messages	list[Any]	The checkpointer's state['messages'] list (LangChain BaseMessage rows), oldest-first.	required
max_turns	int | None	Cap on user+assistant rows replayed (counted as individual rows, not pairs). Defaults to max_entries. The deque's own maxlen is the hard ceiling regardless.	None

Returns:

Type	Description
int	The number of rows replayed into the deque.

Source code in src/symfonic/memory/layers/working.py

async def rehydrate_from_messages(
    self,
    scope: TenantScope,
    messages: list[Any],
    *,
    max_turns: int | None = None,
) -> int:
    """Replay the tail of a persisted transcript into the working deque.

    v7.27.0 restart-resume (Q7, option-b): on session resume the
    in-memory deque is empty (process restarted), but the durable
    checkpointer still holds ``state['messages']``.  This method
    reconstructs the RECENCY WINDOW only -- the last N user/assistant
    turns -- as ``MemoryEntry`` rows tagged with the SAME
    ``speaker``/``turn_id``/``source`` metadata shape the live
    auto-turn writes use (engine.py), so downstream consumers cannot
    tell a rehydrated row from a freshly-written one.

    Scope of restore (LOCKED, plan §4.c): recency window ONLY.  Tool /
    system messages are skipped (they are not conversational turns and
    would blow the cap with noise -- R4).  Durable HMS state is NOT
    touched (it was never lost).  No graph write is made -- this is a
    pure in-memory deque replay; persisting these rows again would
    double-count graph nodes.

    Args:
        scope: Tenant scope selecting the target deque.
        messages: The checkpointer's ``state['messages']`` list
            (LangChain ``BaseMessage`` rows), oldest-first.
        max_turns: Cap on user+assistant rows replayed (counted as
            individual rows, not pairs).  Defaults to ``max_entries``.
            The deque's own ``maxlen`` is the hard ceiling regardless.

    Returns:
        The number of rows replayed into the deque.
    """
    import uuid as _uuid

    cap = self._max if max_turns is None else min(max_turns, self._max)
    if cap <= 0 or not messages:
        return 0

    # Build conversational rows (user/assistant only), oldest-first,
    # then keep only the last ``cap`` so we restore the most recent
    # recency window.
    rows: list[MemoryEntry] = []
    for message in messages:
        speaker = self._speaker_of(message)
        if speaker is None:
            continue  # tool / system / unknown -- not a turn
        content = self._content_of(message)
        if not content:
            continue
        mid = getattr(message, "id", None)
        rows.append(
            MemoryEntry(
                content=content,
                layer=MemoryLayer.WORKING,
                tenant_id=scope.tenant_id,
                metadata={
                    "speaker": speaker,
                    "turn_id": mid or _uuid.uuid4().hex[:12],
                    "source": "resume_rehydrate",
                },
            )
        )

    tail = rows[-cap:]
    key = self._buffer_key(scope)
    buffer = self._buffers[key]
    for entry in tail:
        buffer.append(entry)
    return len(tail)

render_for_prompt

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

Render a working-memory entry for MEMORY_CONTEXT.

v7.7: same speaker-prefix shape as :class:EpisodicLayer -- per-message rows emit [working] <speaker>: <content>; pre- v7.7 rows (no speaker, or speaker == 'legacy') delegate to :func:render_legacy for byte-identical pre-v7.7 output.

v7.7.1: the engine's working-memory pre-hydration block (see SymfonicAgent._hydrate_impl -- the recency-prefix loop immediately preceding the polymorphic-dispatch render) calls this method on each recent entry so the per-layer render contract covers both code paths. Pre-7.7.1 the recency prefix hand-built [working] <content> inline, bypassing the protocol -- a stale docstring previously cited engine.py:5184-5189 for the call site, which was both the wrong line range AND a path that never invoked the renderer (same bypass class as the v7.6.4 procedural-render-bypass).

Source code in src/symfonic/memory/layers/working.py

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

    v7.7: same speaker-prefix shape as :class:`EpisodicLayer` --
    per-message rows emit ``[working] <speaker>: <content>``; pre-
    v7.7 rows (no speaker, or ``speaker == 'legacy'``) delegate
    to :func:`render_legacy` for byte-identical pre-v7.7 output.

    v7.7.1: the engine's working-memory pre-hydration block (see
    ``SymfonicAgent._hydrate_impl`` -- the recency-prefix loop
    immediately preceding the polymorphic-dispatch render) calls
    this method on each recent entry so the per-layer render
    contract covers both code paths.  Pre-7.7.1 the recency
    prefix hand-built ``[working] <content>`` inline, bypassing
    the protocol -- a stale docstring previously cited
    ``engine.py:5184-5189`` for the call site, which was both the
    wrong line range AND a path that never invoked the renderer
    (same bypass class as the v7.6.4 procedural-render-bypass).
    """
    metadata = getattr(entry, "metadata", {}) or {}
    speaker = metadata.get("speaker")
    if isinstance(speaker, str) and speaker and speaker != "legacy":
        content = str(getattr(entry, "content", ""))
        layer_name = getattr(entry, "layer", "working")
        return f"[{layer_name}] {speaker}: {content}"
    return render_legacy(entry, scrubber)

retrieve async

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

Retrieve current working context entries.

v7.22 T-7.22.9: emits :class:MultiScopeIgnoredWarning on first encounter of a scope with inherits_from (plan §2.d wires warning here ONLY -- push / get_context have no scope-merge semantics to ignore, so emitting from them would be a false-positive in adopter mental models). Working memory is a per-(tenant, sub_tenant) FIFO buffer; merging inherited scopes would either leak cross-tenant context or require a merge policy with no defensible default.

Source code in src/symfonic/memory/layers/working.py

async def retrieve(
    self, scope: TenantScope, query: str, top_k: int = 5
) -> list[MemoryEntry]:
    """Retrieve current working context entries.

    v7.22 T-7.22.9: emits :class:`MultiScopeIgnoredWarning` on first
    encounter of a scope with ``inherits_from`` (plan §2.d wires
    warning here ONLY -- ``push`` / ``get_context`` have no
    scope-merge semantics to ignore, so emitting from them would be
    a false-positive in adopter mental models).  Working memory is a
    per-``(tenant, sub_tenant)`` FIFO buffer; merging inherited
    scopes would either leak cross-tenant context or require a
    merge policy with no defensible default.
    """
    from symfonic.memory._multi_scope_warn import maybe_warn_multi_scope

    maybe_warn_multi_scope(self, self._multi_scope_warned, scope)
    entries = await self.get_context(scope)
    return entries[:top_k]

summarize async

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

Summarise working-memory entries according to stm_summary_mode.

v7.0 T09 shipped the extractive summariser but the in-layer summarize() hard-coded mode="extractive" regardless of config. v7.0.1 R3 wires the stm_summary_mode kwarg through so direct-API callers see the same behaviour as engine-routed callers:

• "off" (default) — returns "". Preserves v6.1.x byte-for-byte behaviour.
• "extractive" — delegates to summarize_entries with the deterministic first-sentence / last-sentence heuristic. Zero-cost, no LLM call.
• "llm" — the in-layer summarize() is synchronous- equivalent over the MemoryEntry list and does not own a model provider. Falls through to the extractive path so this call never stalls on an upstream outage; callers who want the true LLM micro-call must invoke :func:symfonic.memory.working.stm_summary.summarize_entries directly with a model_provider (the engine wires that path via SymfonicAgent).

Empty input always returns the empty string so the v6.1.x observable "no meaningful summary" behaviour is preserved on empty buffers independent of mode.

Source code in src/symfonic/memory/layers/working.py

async def summarize(
    self,
    scope: TenantScope,
    entries: list[MemoryEntry],
) -> str:
    """Summarise working-memory entries according to ``stm_summary_mode``.

    v7.0 T09 shipped the extractive summariser but the in-layer
    ``summarize()`` hard-coded ``mode="extractive"`` regardless of
    config. v7.0.1 R3 wires the ``stm_summary_mode`` kwarg through
    so direct-API callers see the same behaviour as engine-routed
    callers:

    - ``"off"`` (default) — returns ``""``. Preserves v6.1.x
      byte-for-byte behaviour.
    - ``"extractive"`` — delegates to ``summarize_entries`` with
      the deterministic first-sentence / last-sentence heuristic.
      Zero-cost, no LLM call.
    - ``"llm"`` — the in-layer ``summarize()`` is synchronous-
      equivalent over the MemoryEntry list and does not own a
      model provider. Falls through to the extractive path so
      this call never stalls on an upstream outage; callers who
      want the true LLM micro-call must invoke
      :func:`symfonic.memory.working.stm_summary.summarize_entries`
      directly with a ``model_provider`` (the engine wires that
      path via ``SymfonicAgent``).

    Empty input always returns the empty string so the v6.1.x
    observable "no meaningful summary" behaviour is preserved on
    empty buffers independent of mode.
    """
    if not entries:
        return ""
    if self._stm_summary_mode == "off":
        return ""
    # Both "extractive" and "llm" route through the extractive
    # path here; LLM-mode requires the async summarize_entries
    # call with a model_provider (callers must invoke directly).
    from symfonic.memory.working.stm_summary import summarize_entries

    return summarize_entries(entries, mode="extractive").text

write async

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

Push an entry to the working buffer (FIFO eviction) and persist to graph.

Source code in src/symfonic/memory/layers/working.py

async def write(self, scope: TenantScope, entry: MemoryEntry) -> None:
    """Push an entry to the working buffer (FIFO eviction) and persist to graph."""
    await self.push(scope, entry)