symfonic.agent.history_trim

history_trim

Conversation-history trimming shared by the legacy and kernel routes.

Extracted verbatim from symfonic.agent.engine so the migrated kernel route (:func:symfonic.agent.cutover.messages.as_kernel_history) can trim without importing the engine module. engine retains lazy wrappers under the original private names, so the legacy body and every existing import of symfonic.agent.engine._pair_aware_history_slice / _log_history_trim keep working through the same implementations.

The logger name is "symfonic.agent" -- the engine's own hardcoded logger name -- so the DEBUG record keeps its identity for operators filtering on it.

log_history_trim

log_history_trim(history: Sequence[Any] | None, cap: int) -> None

DEBUG-log how many oldest history messages cap is about to drop.

Added in v7.1.2 so operator-side "memory loss" reports can be told apart from retrieval/consolidation losses. Module-level rather than a method because both invocation routes have to emit it: the legacy body calls it through :meth:SymfonicAgent._maybe_log_history_trim, and the migrated kernel route calls it from :func:symfonic.agent.cutover.messages.as_kernel_history. A route that trims silently while the other one narrates the same trim is an observable difference on the one diagnostic that exists for this class of report.

Source code in src/symfonic/agent/history_trim.py

def log_history_trim(history: Sequence[Any] | None, cap: int) -> None:
    """DEBUG-log how many oldest history messages ``cap`` is about to drop.

    Added in v7.1.2 so operator-side "memory loss" reports can be told apart
    from retrieval/consolidation losses. Module-level rather than a method
    because *both* invocation routes have to emit it: the legacy body calls it
    through :meth:`SymfonicAgent._maybe_log_history_trim`, and the migrated
    kernel route calls it from
    :func:`symfonic.agent.cutover.messages.as_kernel_history`. A route that
    trims silently while the other one narrates the same trim is an observable
    difference on the one diagnostic that exists for this class of report.
    """
    if history is None:
        return
    if len(history) > cap:
        logger.debug(
            "Dropping %d oldest history messages (cap=%d). To diagnose "
            "memory-loss reports, raise agent.max_conversation_messages or "
            "ensure your hydration layers cover the dropped range.",
            len(history) - cap, cap,
        )

pair_aware_history_slice

pair_aware_history_slice(history: list[Any], cap: int) -> list[Any]

v7.12.0: cap a conversation-history list to the most recent cap entries WITHOUT cutting between an AIMessage's tool_use block and its matching ToolMessage.

Background. Anthropic's API rejects (HTTP 400) any message list where an AIMessage carries a tool_use block without a following tool_result block with the matching tool_use_id. The pre-v7.12.0 slice history[-cap:] was blunt -- it would happily cut between an AIMessage and its ToolMessage, leaving the leading message dangling and producing a 400 on the next call. With v7.12.0 tool-result compaction shrinking individual ToolMessages, more messages fit under the cap and this latent bug surfaced more often.

Algorithm. Walk from the end of history backwards. Keep messages until the count >= cap. After hitting cap, continue walking back through any AIMessage whose tool_calls are not all paired by ToolMessages already in the retained slice. This guarantees every retained AIMessage's tool_use blocks have a matching ToolMessage in the slice.

Defensive: returns history[-cap:] unchanged when cap >= len(history) (no slicing needed) and when cap <= 0 (returns empty list, matching the pre-existing semantics).

Source code in src/symfonic/agent/history_trim.py

def pair_aware_history_slice(
    history: list[Any], cap: int,
) -> list[Any]:
    """v7.12.0: cap a conversation-history list to the most recent
    ``cap`` entries WITHOUT cutting between an AIMessage's
    ``tool_use`` block and its matching ``ToolMessage``.

    Background.  Anthropic's API rejects (HTTP 400) any message list
    where an ``AIMessage`` carries a ``tool_use`` block without a
    following ``tool_result`` block with the matching
    ``tool_use_id``.  The pre-v7.12.0 slice ``history[-cap:]`` was
    blunt -- it would happily cut between an AIMessage and its
    ToolMessage, leaving the leading message dangling and producing
    a 400 on the next call.  With v7.12.0 tool-result compaction
    shrinking individual ToolMessages, more messages fit under the
    cap and this latent bug surfaced more often.

    Algorithm.  Walk from the end of ``history`` backwards.  Keep
    messages until the count >= cap.  After hitting cap, continue
    walking back through any AIMessage whose ``tool_calls`` are not
    all paired by ToolMessages already in the retained slice.  This
    guarantees every retained AIMessage's tool_use blocks have a
    matching ToolMessage in the slice.

    Defensive: returns ``history[-cap:]`` unchanged when ``cap >=
    len(history)`` (no slicing needed) and when ``cap <= 0``
    (returns empty list, matching the pre-existing semantics).
    """
    if cap <= 0:
        return []
    # Step 1: start with the naive tail-slice as the seed.  When
    # ``cap >= len(history)`` this is the full list, but we still
    # need to run Step 5 to drop orphaned AIMessages whose matching
    # ToolMessages were upstream-truncated -- the input might already
    # be malformed.
    retained = list(history if cap >= len(history) else history[-cap:])
    retained_ids = {id(m) for m in retained}
    # Step 2: collect the set of tool_call_ids that the retained
    # slice carries as ToolMessages (the "satisfied" set).
    satisfied: set[str] = set()
    for m in retained:
        tcid = getattr(m, "tool_call_id", None)
        if tcid:
            satisfied.add(str(tcid))
    # Step 3: collect the set of tool_call_ids the retained slice
    # REQUIRES via AIMessage.tool_calls but does NOT have ToolMessages
    # for.  These are the dangling refs we need to either (a) extend
    # backwards to pick up their ToolMessages, or (b) drop the
    # AIMessage that requires them.
    def _required_ids(messages_list: list[Any]) -> set[str]:
        needed: set[str] = set()
        for msg in messages_list:
            tool_calls = getattr(msg, "tool_calls", None)
            if not isinstance(tool_calls, list):
                continue
            for tc in tool_calls:
                tid = (
                    tc.get("id") if isinstance(tc, dict)
                    else getattr(tc, "id", None)
                )
                if tid:
                    needed.add(str(tid))
        return needed

    # Step 4: walk further back to pull in matching ToolMessages for
    # any dangling required ids.  Bounded by the size of `history`
    # so it cannot loop forever.
    cut_index = len(history) - cap  # first index NOT yet in retained
    while True:
        required = _required_ids(retained)
        missing = required - satisfied
        if not missing or cut_index <= 0:
            break
        cut_index -= 1
        candidate = history[cut_index]
        # Only pull this message in if it satisfies one of the missing
        # ids; this prevents reverse-walking through irrelevant
        # messages just to satisfy a near-orphan.
        cand_tcid = str(getattr(candidate, "tool_call_id", "") or "")
        if cand_tcid in missing:
            if id(candidate) not in retained_ids:
                retained.insert(0, candidate)
                retained_ids.add(id(candidate))
                satisfied.add(cand_tcid)
        else:
            # Stop the walk-back the moment we hit a message that
            # doesn't satisfy a dangling id -- otherwise we'd grow
            # the slice past the user's intended cap chasing edges.
            break

    # Step 5: drop any AIMessage at the FRONT of the retained slice
    # whose tool_calls still aren't all satisfied (defensive cleanup
    # for the case where the matching ToolMessages weren't in
    # ``history`` at all -- e.g. truncated upstream).
    while retained:
        head = retained[0]
        head_tool_calls = getattr(head, "tool_calls", None)
        if not isinstance(head_tool_calls, list) or not head_tool_calls:
            break
        head_ids = {
            str(tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", ""))
            for tc in head_tool_calls
        }
        head_ids.discard("")
        if head_ids.issubset(satisfied):
            break
        # Head AIMessage requires ToolMessages we don't have; drop it.
        retained.pop(0)

    # Step 6 (v8.7.2): drop any leading orphan ToolMessage.  A ToolMessage
    # at the FRONT of the slice has no preceding AIMessage ``tool_use`` in
    # the retained list -- Anthropic rejects such a ``tool_result`` with an
    # HTTP 400 the same way it rejects a dangling ``tool_use``.  This is the
    # mirror of Step 5 and closes the front-side hole on checkpointed
    # threads whose owning AIMessage was truncated upstream.
    while retained and getattr(retained[0], "tool_call_id", None):
        retained.pop(0)
    return retained