symfonic.core.nodes.context_window

context_window

ContextWindowNode -- conversation summarization when token threshold exceeded.

Per TDD 2.12: All thresholds come from CompactionConfig. Zero hardcoded constants inside the node body.

Renamed from CompactionNode to better reflect its purpose: this node manages the LangGraph context window, not memory-layer compaction.

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 Jarvio 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.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}"

create_context_window_node

create_context_window_node(
    config: CompactionConfig,
    *,
    model_config: ModelConfig | None = None,
    role_models: dict[str, ModelConfig] | None = None,
) -> Callable[..., Any]

Factory for ContextWindowNode. Config-driven, no hardcoded constants.

Parameters:

Name	Type	Description	Default
config	CompactionConfig	Compaction parameters (budget knobs, summarization model, keep-recent count).	required
model_config	ModelConfig | None	Running agent's model configuration. When provided, model_config.context_window participates in the fallback chain for the hard budget so non-Claude models are not silently pinned to Claude's 200k tokens. Optional for back-compat: omitting it preserves the v7.1.1 behaviour (200_000 safety net).	None
role_models	dict[str, ModelConfig] | None	FrameworkConfig.role_models snapshot at compile time. When role_models.get("summary") is a :class:ModelConfig, the summariser dispatches through it instead of constructing a fresh ModelConfig from CompactionConfig.compaction_model. v7.24.2 closure of the documented-but-unwired summary role contract (see agent/config.py:1050-1051). None or absent "summary" key preserves byte-identical pre-v7.24.2 behaviour.	None

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

def create_context_window_node(
    config: CompactionConfig,
    *,
    model_config: ModelConfig | None = None,
    role_models: dict[str, ModelConfig] | None = None,
) -> Callable[..., Any]:
    """Factory for ContextWindowNode. Config-driven, no hardcoded constants.

    Args:
        config: Compaction parameters (budget knobs, summarization model,
            keep-recent count).
        model_config: Running agent's model configuration. When provided,
            ``model_config.context_window`` participates in the fallback
            chain for the hard budget so non-Claude models are not
            silently pinned to Claude's 200k tokens. Optional for
            back-compat: omitting it preserves the v7.1.1 behaviour
            (200_000 safety net).
        role_models: ``FrameworkConfig.role_models`` snapshot at compile
            time.  When ``role_models.get("summary")`` is a
            :class:`ModelConfig`, the summariser dispatches through it
            instead of constructing a fresh ``ModelConfig`` from
            ``CompactionConfig.compaction_model``.  v7.24.2 closure of
            the documented-but-unwired ``summary`` role contract
            (see ``agent/config.py:1050-1051``).  ``None`` or absent
            ``"summary"`` key preserves byte-identical pre-v7.24.2
            behaviour.
    """

    @observed_node("compaction", swallow_errors=True)
    async def context_window(state: dict[str, Any]) -> dict[str, Any]:
        from ..callbacks.emit import emit_llm_end_from_result
        from ..callbacks.manager import CallbackManager
        from ..observability.protocol import ObservabilityHook
        from ..providers import ModelProvider

        deps = state["deps"]

        messages = list(state["messages"])
        total_chars = _estimate_chars(messages)
        budget = _compute_budget_chars(config, model_config)

        # Soft threshold: compact early when estimated tokens exceed
        # soft_threshold_tokens, even before the hard budget is exhausted.
        soft_exceeded = _exceeds_soft_threshold(messages, config)
        if total_chars <= budget and not soft_exceeded:
            return {}

        provider = deps.require(ModelProvider)
        compaction_config_model = config.compaction_model

        # Keep recent messages intact
        keep_count = config.keep_recent_messages
        if len(messages) <= keep_count:
            return {}

        # keep_count == 0 means "keep nothing recent" -> compact everything
        # (``messages[:-0]`` is ``[]``, a silent no-op burn).
        split = len(messages) - keep_count if keep_count > 0 else len(messages)
        # Pair-aware boundary: the kept (recent) region must not BEGIN with a
        # ToolMessage whose owning AIMessage is being removed -- that orphaned
        # tool_result triggers an Anthropic 400. Advance the split forward past
        # any leading ToolMessages so the kept region starts at a clean
        # AIMessage/HumanMessage boundary (the orphans are summarized + removed).
        while split < len(messages) and isinstance(messages[split], ToolMessage):
            split += 1
        old_messages = messages[:split]
        summary_text = "\n".join(
            f"{type(m).__name__}: {m.content}" for m in old_messages
        )

        from ..config import ModelConfig

        # v7.24.2: honour FrameworkConfig.role_models["summary"] when
        # the adopter has pinned a summariser model.  Falls back to the
        # CompactionConfig.compaction_model literal (pre-v7.24.2
        # behaviour) when no override is set.  The override is taken
        # as-is -- max_tokens / temperature / extra_headers etc. are
        # whatever the adopter declared.  See
        # ``agent/config.py:1050-1051`` for the documented contract.
        _summary_override = (role_models or {}).get("summary")
        if isinstance(_summary_override, ModelConfig):
            summary_config = _summary_override
        else:
            summary_config = ModelConfig(
                model_name=compaction_config_model,
                max_tokens=config.max_chunk_tokens,
            )
        llm = provider.get_chat_model(summary_config)

        summary_prompt = (
            "Summarize the following conversation history concisely, "
            "preserving key decisions, facts, and context:\n\n"
            f"{summary_text}"
        )

        # v7.15.0: capture wall-clock + UTC start via the canonical
        # context manager so LLMEndEvent carries duration_ms /
        # started_at_utc.
        from ..callbacks.emit import llm_timing, resolve_model_name

        # v8.3.2 observability-bug fix (G1): the summariser LLM call was
        # INVISIBLE to the always-on ``ObservabilityHook`` path (cortex/OTel
        # spans) because it emitted ONLY on the merge-dependent callback
        # manager -- the exact bug class the v8.3.1 critic fix closed.  We
        # now mirror react (``react.py:1078``/``:1433``): emit
        # ``on_llm_start``/``on_llm_end`` on the always-on hook, which is
        # reachable here via ``deps.require(ObservabilityHook)`` (defaults
        # to ``NoOpObservabilityHook`` -> byte-identical when unwired).
        # ATTRIBUTION is MODEL-BASED (the hook signature carries no
        # node_name); cortex distinguishes the summariser from react/critic
        # by the resolved ``summary``-role SKU.  Node-level disambiguation
        # via run_id-join to the callback LLMEndEvent (which DOES carry
        # node_name) is deferred to Build 2.
        _hook = deps.require(ObservabilityHook)
        _summary_model_name = resolve_model_name(llm, summary_config.model_name)
        _summary_messages = [HumanMessage(content=summary_prompt)]
        _run_id = state.get("run_id", "") or ""

        try:
            await _hook.on_llm_start(
                _summary_model_name, _summary_messages, _run_id
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_start failed")

        # v8.4.0 CALLBACK-PATH fix: open the OTel span on the merged
        # manager BEFORE the ainvoke (the END at :289 closes it).  Resolve
        # the manager here (was resolved after the call pre-v8.4.0) so the
        # start/end pair share it.  Same node_name as the END; byte-identical
        # no-op when no manager is merged.
        callback_mgr: CallbackManager | None = (
            state.get("_callback_manager") or deps.get(CallbackManager)
        )
        if callback_mgr is not None and not callback_mgr.is_noop:
            from ..callbacks.emit import emit_llm_start

            await emit_llm_start(
                callback_mgr,
                model=_summary_model_name,
                messages=_summary_messages,
                run_id=_run_id,
                node_name="context_window_summariser",
            )

        async with llm_timing() as _timing:
            summary_response = await llm.ainvoke(_summary_messages)

        try:
            from ..callbacks.emit import _extract_usage

            await _hook.on_llm_end(
                _summary_model_name,
                str(getattr(summary_response, "content", summary_response)),
                _extract_usage(summary_response),
                _run_id,
            )
        except Exception:
            logger.exception("ObservabilityHook.on_llm_end failed")

        # v7.4.3 Jarvio Ask 5: emit on_llm_end so the ALWAYS-ON
        # summarisation call is visible to token billing / OTel.
        # ``callback_mgr`` was resolved BEFORE the ainvoke (v8.4.0) so the
        # start/end pair share the same manager.
        # v7.14.0: stamp the resolved SKU (what the provider actually
        # returned) instead of the requested compaction_model.  Falls back
        # to summary_config.model_name when the chat model doesn't expose
        # ``.model`` / ``.model_name``.  See callbacks/emit.py docstring.
        await emit_llm_end_from_result(
            callback_mgr,
            model=_summary_model_name,
            result=summary_response,
            run_id=_run_id,
            node_name="context_window_summariser",
            timing=_timing,
        )

        # Build removal messages for old messages + inject summary
        removals: list[RemoveMessage] = []
        for m in old_messages:
            if hasattr(m, "id") and m.id:
                removals.append(RemoveMessage(id=m.id))

        # v7.25.0 T5: honesty-marker sentinel.  ``last_compacted_index``
        # is the count of messages the summariser just removed -- a
        # positive value means compaction fired this turn.  After the
        # removal, the model sees only the kept-recent slice; we treat
        # the "no retrieval came back" case as ``entries_count=0`` so
        # the cross-session fabrication failure class (T14) cannot
        # silently invite the model to confabulate from another session.
        # The kept-recent slice itself is NOT counted toward
        # ``entries_count`` because it lives in the LangGraph message
        # array, not the MEMORY_CONTEXT layer the model treats as
        # historical recall.  See ``compose_memory_section``'s contract.
        summary_body = f"[Conversation Summary]\n{summary_response.content}"
        memory_section = compose_memory_section(
            summary_text=summary_body,
            last_compacted_index=len(old_messages),
            entries_count=0,
        )

        result_messages: list[BaseMessage] = [
            *removals,
            SystemMessage(content=memory_section),
        ]

        result: dict[str, Any] = {
            "messages": result_messages,
        }

        # Memory flush -- separate concern, delegated to flush_memory()
        if config.enable_memory_flush:
            await flush_memory(state)

        return result

    return context_window

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.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)