LCORE-1570: utils/compaction: recursive resummarize fallback

Add recursively_resummarize — the spec's escape hatch when additive
compaction's per-chunk summaries themselves grow large enough to
threaten the context window. Without it, the additive design grows
its summary set unboundedly across a long-running conversation; the
fold collapses every chunk into a single ConversationSummary.

The function takes a list of existing summaries (oldest first) and
calls the LLM once with a prompt that lists each chunk in order and
asks for a combined summary preserving the same five Red Hat
support directives. The new summary inherits summarized_through_turn
from the most recent input — no new conversation turns were
summarized by this call; the running total has not advanced.

Folding a single summary is a no-op and raises ValueError so a
caller that only has one chunk does not silently make an unnecessary
LLM call. Same for an empty list.

A separate RECURSIVE_RESUMMARIZATION_PROMPT constant keeps the
fold's prompt distinct from the additive prompt (the input is
qualitatively different — already-summarized text rather than raw
transcript — though the preservation directives are the same).

The function intentionally does not decide *when* to fold. The
caller (LCORE-1572) compares the cumulative summary token count
against some configured fraction of the context window and invokes
this only at the threshold.

Tests cover:
* Multi-summary fold returns one ConversationSummary inheriting the
  most recent through-turn count.
* Prompt body contains every input summary's text and the fallback
  prompt constant.
* store=False / stream=False on the LLM call.
* Single-summary and empty-list inputs raise ValueError.
* Empty LLM response raises ValueError (mirrors summarize_chunk).

Author: max-svistunov

src/utils/compaction.py

@@ -354,3 +354,111 @@ async def summarize_chunk(
         created_at=datetime.now(timezone.utc).isoformat(),
         model_used=model,
     )
+
+
+RECURSIVE_RESUMMARIZATION_PROMPT = (
+    "The following are summaries of older portions of a conversation"
+    " between a user and an AI assistant helping with Red Hat product"
+    " support. Combine them into a single summary that preserves the"
+    " same five categories of detail (original question and"
+    " environment, error messages and outcomes, decisions and"
+    " rationale, what was resolved versus what is open, and clear"
+    " attribution between user and assistant). Resolve repetition;"
+    " keep all distinct facts.\n"
+)
+"""Prompt used when collapsing multiple summaries into one.
+
+Separate from SUMMARIZATION_PROMPT because the input is qualitatively
+different (already-summarized text, not raw conversation), but the
+preservation directives are the same.
+"""
+
+
+async def recursively_resummarize(
+    client: AsyncLlamaStackClient,
+    model: str,
+    summaries: list[ConversationSummary],
+    encoding_name: str,
+) -> ConversationSummary:
+    """Collapse multiple ``ConversationSummary`` records into one.
+
+    This is the fallback for the additive design (spike decision 2).
+    Each compaction normally produces a new summary chunk that is kept
+    alongside the previous ones, but the spec mandates a recursive
+    fold-up when the cumulative size of the summaries themselves
+    approaches the context limit:
+
+        When total summary token count itself approaches the context
+        limit, fall back to recursive re-summarization of the oldest
+        summary chunks.
+        — docs/design/conversation-compaction/conversation-compaction.md
+
+    The caller decides *when* to invoke this (typically after measuring
+    that the total summary tokens cross some configured fraction of
+    the model's context window). This function carries out the fold:
+    it builds a prompt that lists each existing summary in order, asks
+    the LLM to produce a single combined summary preserving the same
+    five directives, and returns a fresh ``ConversationSummary``.
+
+    The returned summary inherits ``summarized_through_turn`` from the
+    most recent input summary (the running total has not advanced —
+    we have re-folded, not summarized anything new).
+
+    Parameters:
+        client: Llama Stack client to call.
+        model: Fully-qualified model identifier used for the LLM call.
+        summaries: Existing summary chunks to fold, in chronological
+            order (oldest first). Must contain at least two entries —
+            folding a single summary is a no-op the caller should
+            short-circuit before invoking this function.
+        encoding_name: Tiktoken encoding name used to count tokens in
+            the produced fold.
+
+    Returns:
+        A single ConversationSummary representing the union of
+        ``summaries``.
+
+    Raises:
+        ValueError: When *summaries* has fewer than two entries (no
+            fold is needed) or when the LLM call yields no
+            extractable text.
+    """
+    from utils.token_estimator import estimate_tokens
+
+    if len(summaries) < 2:
+        raise ValueError(
+            "recursively_resummarize requires at least 2 summary chunks "
+            "to fold; caller must short-circuit when fewer are present."
+        )
+
+    transcript = "\n\n".join(
+        f"Summary {i + 1} (through turn {s.summarized_through_turn}):\n"
+        f"{s.summary_text}"
+        for i, s in enumerate(summaries)
+    )
+    prompt = f"{RECURSIVE_RESUMMARIZATION_PROMPT}\n{transcript}"
+
+    logger.info(
+        "Recursively re-summarizing %d existing summary chunks for model %s.",
+        len(summaries),
+        model,
+    )
+    response = await client.responses.create(
+        input=prompt,
+        model=model,
+        stream=False,
+        store=False,
+    )
+    folded_text = _extract_response_text(response).strip()
+    if not folded_text:
+        raise ValueError(
+            "Recursive re-summarization LLM call returned no extractable "
+            "text; cannot fold the existing ConversationSummary records."
+        )
+    return ConversationSummary(
+        summary_text=folded_text,
+        summarized_through_turn=summaries[-1].summarized_through_turn,
+        token_count=estimate_tokens(folded_text, encoding_name=encoding_name),
+        created_at=datetime.now(timezone.utc).isoformat(),
+        model_used=model,
+    )

tests/unit/utils/test_compaction.py

@@ -5,14 +5,17 @@
 import pytest
 from pytest_mock import MockerFixture
 
+from models.compaction import ConversationSummary
 from utils.compaction import (
+    RECURSIVE_RESUMMARIZATION_PROMPT,
     SUMMARIZATION_PROMPT,
     _build_summarization_prompt_body,
     _extract_response_text,
     extract_message_text,
     format_conversation_for_summary,
     is_message_item,
     partition_conversation,
+    recursively_resummarize,
     summarize_chunk,
 )
 from utils.token_estimator import (
@@ -499,3 +502,145 @@ async def test_additive_second_call_does_not_resummarize_first(
         assert first.summary_text == "First summary."
         assert second.summary_text == "Second summary."
         assert second.summarized_through_turn > first.summarized_through_turn
+
+
+# ---------------------------------------------------------------------------
+# recursively_resummarize
+# ---------------------------------------------------------------------------
+
+
+def _make_summary(text: str, through: int, tokens: int = 5) -> ConversationSummary:
+    """Build a ConversationSummary value for tests."""
+    return ConversationSummary(
+        summary_text=text,
+        summarized_through_turn=through,
+        token_count=tokens,
+        created_at="2026-05-11T00:00:00Z",
+        model_used="openai/gpt-4o-mini",
+    )
+
+
+class TestRecursivelyResummarize:
+    """Tests for the recursive-fold fallback."""
+
+    @pytest.mark.asyncio
+    async def test_collapses_n_summaries_into_one(
+        self, mocker: MockerFixture
+    ) -> None:
+        """Multiple summaries fold into one ConversationSummary."""
+        client = mocker.AsyncMock()
+        client.responses.create.return_value = _make_summary_response(
+            mocker, "Combined summary covering all prior chunks."
+        )
+        summaries = [
+            _make_summary("First chunk summary.", through=5),
+            _make_summary("Second chunk summary.", through=10),
+            _make_summary("Third chunk summary.", through=15),
+        ]
+        folded = await recursively_resummarize(
+            client=client,
+            model="openai/gpt-4o-mini",
+            summaries=summaries,
+            encoding_name=DEFAULT_ENCODING_NAME,
+        )
+        assert (
+            folded.summary_text
+            == "Combined summary covering all prior chunks."
+        )
+        # The fold inherits the most recent input's running total — no
+        # new turns were summarized by this call.
+        assert folded.summarized_through_turn == 15
+        assert folded.token_count > 0
+        assert folded.model_used == "openai/gpt-4o-mini"
+
+    @pytest.mark.asyncio
+    async def test_prompt_lists_each_summary(
+        self, mocker: MockerFixture
+    ) -> None:
+        """All input summary texts and the fallback prompt appear in the call."""
+        client = mocker.AsyncMock()
+        client.responses.create.return_value = _make_summary_response(
+            mocker, "Folded summary."
+        )
+        summaries = [
+            _make_summary("Alpha facts.", through=5),
+            _make_summary("Beta facts.", through=10),
+        ]
+        await recursively_resummarize(
+            client=client,
+            model="openai/gpt-4o-mini",
+            summaries=summaries,
+            encoding_name=DEFAULT_ENCODING_NAME,
+        )
+        prompt_input = client.responses.create.call_args.kwargs["input"]
+        assert RECURSIVE_RESUMMARIZATION_PROMPT in prompt_input
+        assert "Alpha facts." in prompt_input
+        assert "Beta facts." in prompt_input
+        assert "Summary 1" in prompt_input
+        assert "Summary 2" in prompt_input
+
+    @pytest.mark.asyncio
+    async def test_uses_store_false(self, mocker: MockerFixture) -> None:
+        """The recursive call also uses store=False — like the additive one."""
+        client = mocker.AsyncMock()
+        client.responses.create.return_value = _make_summary_response(
+            mocker, "Folded."
+        )
+        summaries = [
+            _make_summary("a", through=1),
+            _make_summary("b", through=2),
+        ]
+        await recursively_resummarize(
+            client=client,
+            model="openai/gpt-4o-mini",
+            summaries=summaries,
+            encoding_name=DEFAULT_ENCODING_NAME,
+        )
+        kwargs = client.responses.create.call_args.kwargs
+        assert kwargs["store"] is False
+        assert kwargs["stream"] is False
+
+    @pytest.mark.asyncio
+    async def test_raises_for_single_summary(
+        self, mocker: MockerFixture
+    ) -> None:
+        """Folding a single summary is a no-op the caller must avoid."""
+        client = mocker.AsyncMock()
+        with pytest.raises(ValueError, match="at least 2 summary chunks"):
+            await recursively_resummarize(
+                client=client,
+                model="openai/gpt-4o-mini",
+                summaries=[_make_summary("only one", through=5)],
+                encoding_name=DEFAULT_ENCODING_NAME,
+            )
+
+    @pytest.mark.asyncio
+    async def test_raises_for_empty_list(self, mocker: MockerFixture) -> None:
+        """Folding an empty list is a no-op the caller must avoid."""
+        client = mocker.AsyncMock()
+        with pytest.raises(ValueError, match="at least 2 summary chunks"):
+            await recursively_resummarize(
+                client=client,
+                model="openai/gpt-4o-mini",
+                summaries=[],
+                encoding_name=DEFAULT_ENCODING_NAME,
+            )
+
+    @pytest.mark.asyncio
+    async def test_raises_when_llm_returns_empty(
+        self, mocker: MockerFixture
+    ) -> None:
+        """Empty LLM response surfaces ValueError, not an empty fold."""
+        client = mocker.AsyncMock()
+        client.responses.create.return_value = _make_summary_response(mocker, "")
+        summaries = [
+            _make_summary("a", through=1),
+            _make_summary("b", through=2),
+        ]
+        with pytest.raises(ValueError, match="no extractable text"):
+            await recursively_resummarize(
+                client=client,
+                model="openai/gpt-4o-mini",
+                summaries=summaries,
+                encoding_name=DEFAULT_ENCODING_NAME,
+            )