symfonic.core.nodes.context_window_memory

context_window_memory

Pure helpers behind the context-window node: budget, threshold, memory.

Split out of :mod:symfonic.core.nodes.context_window (374 lines against the 300-line budget). Everything here is side-effect free apart from flush_memory, which is the one delegation the node makes outwards; the node factory itself stays in context_window.

context_window re-exports every name below, so from symfonic.core.nodes.context_window import compose_memory_section and its siblings -- the form the summariser tests use -- are unchanged.

HONESTY_MARKER_SENTINEL module-attribute

HONESTY_MARKER_SENTINEL: str = "[memory] Earlier turns exist in this conversation's history but are not currently in context."

Load-bearing exact string literal -- adopters MAY pattern-match on this to surface a UI indicator that the model is operating without access to prior turns. Do not change without a deprecation cycle.

Injected at the head of the summariser's memory section when the last_compacted_index > 0 AND entries_count == 0 precondition holds: compaction has fired (prior turns existed) AND retrieval came back empty (no visible context for the model). Closes the cross-session fabrication failure class (T14 in the adopter bench) by giving the model an explicit signal that there ARE earlier turns instead of silently inviting it to confabulate from another session's data.

No LLM call -- this is purely a deterministic guard against the failure class on the current architecture. The structural v8.0 TenantScope.session_id work still lands later.

compose_memory_section

compose_memory_section(*, summary_text: str, last_compacted_index: int, entries_count: int) -> str

Build the summariser's system-side memory section with the v7.25.0 honesty-marker sentinel injected at the head when warranted.

Pure function -- no I/O, no side effects -- so adopters can unit-test against the contract directly and tests can drive the conditions without spinning up the LangGraph node fixture.

Parameters:

Name	Type	Description	Default
summary_text	str	The summariser's existing memory content (typically the compacted [Conversation Summary] body). May be the empty string when compaction produced nothing (e.g. a soft-threshold trip without entries to summarise).	required
last_compacted_index	int	Position (>0 when compaction has fired in this session, 0 when it has not). The semantic anchor is "has the deque ever been emptied by compaction" -- a non-zero value means earlier turns existed and are gone.	required
entries_count	int	Number of entries CURRENTLY visible to the model after retrieval. Zero means the model is staring at no prior conversational context.	required

Returns:

Type	Description
str	"{HONESTY_MARKER_SENTINEL}\n{summary_text}" when the marker
str	precondition (last_compacted_index > 0 AND entries_count == 0)
str	holds. Otherwise summary_text is returned unchanged.

Idempotence: when summary_text already begins with the sentinel, the function MUST NOT prepend a second copy -- adopter pattern-match logic would see two consecutive sentinels and the de-dupe contract would break.

Source code in src/symfonic/core/nodes/context_window_memory.py

def compose_memory_section(
    *,
    summary_text: str,
    last_compacted_index: int,
    entries_count: int,
) -> str:
    """Build the summariser's system-side memory section with the v7.25.0
    honesty-marker sentinel injected at the head when warranted.

    Pure function -- no I/O, no side effects -- so adopters can unit-test
    against the contract directly and tests can drive the conditions
    without spinning up the LangGraph node fixture.

    Args:
        summary_text: The summariser's existing memory content (typically
            the compacted ``[Conversation Summary]`` body).  May be the
            empty string when compaction produced nothing (e.g. a
            soft-threshold trip without entries to summarise).
        last_compacted_index: Position (>0 when compaction has fired in
            this session, 0 when it has not).  The semantic anchor is
            "has the deque ever been emptied by compaction" -- a non-zero
            value means earlier turns existed and are gone.
        entries_count: Number of entries CURRENTLY visible to the model
            after retrieval.  Zero means the model is staring at no prior
            conversational context.

    Returns:
        ``"{HONESTY_MARKER_SENTINEL}\\n{summary_text}"`` when the marker
        precondition (``last_compacted_index > 0 AND entries_count == 0``)
        holds.  Otherwise ``summary_text`` is returned unchanged.

    Idempotence: when ``summary_text`` already begins with the sentinel,
    the function MUST NOT prepend a second copy -- adopter pattern-match
    logic would see two consecutive sentinels and the de-dupe contract
    would break.
    """
    if last_compacted_index <= 0 or entries_count != 0:
        return summary_text

    # Idempotence guard -- if the summary already carries the sentinel
    # (e.g. a callerpre-baked it), don't prepend again.
    if HONESTY_MARKER_SENTINEL in summary_text:
        return summary_text

    if not summary_text:
        return HONESTY_MARKER_SENTINEL

    return f"{HONESTY_MARKER_SENTINEL}\n{summary_text}"

flush_memory async

flush_memory(state: dict[str, Any]) -> None

Flush compacted memory to an external store if configured.

Extracted from the context window node to honour SRP -- context window handles history management only; memory persistence is a separate concern.

Source code in src/symfonic/core/nodes/context_window_memory.py

async def flush_memory(state: dict[str, Any]) -> None:
    """Flush compacted memory to an external store if configured.

    Extracted from the context window node to honour SRP -- context window
    handles history management only; memory persistence is a separate concern.
    """
    from ..deps import BaseAgentDeps
    from ..protocols import MemoryFlusher

    deps: BaseAgentDeps = state["deps"]
    flush_fn = (
        deps.get(MemoryFlusher)
        if isinstance(deps, BaseAgentDeps)
        else None
    )
    if flush_fn and callable(getattr(flush_fn, "flush", None)):
        await flush_fn.flush(state)