symfonic.memory.layers.working

working

WorkingLayer -- bounded active conversation context.

Session-scoped FIFO buffer backed by an optional GraphBackend for persistence. When a graph_store is provided, writes are mirrored to the graph so they survive across process restarts and can be counted by the metrics endpoint. The in-memory buffer is still used for fast within-session retrieval. Evicts oldest in-memory entries when the buffer exceeds max_entries.

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)),
            )
            # TA8.74. A caller-supplied id survives here now, as it always
            # did in the episodic layer. Without it every write minted a
            # fresh node, so a write port's documented upsert became an
            # insert and one turn's exchange landed twice -- measured, four
            # durable rows with four ids for two records.
            if entry.node_id:
                node.id = entry.node_id
            elif entry.id:
                node.id = entry.id
            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)