fix(compression): truncated summaries no longer become compaction checkpoints (port of earendil-works/pi#7048)

A summarization response with finish_reason == "length" contains PARTIAL
text — the generation stopped on the output-token cap mid-summary.
Previously all compressor summarization sites accepted such responses as
complete: the cut-off text replaced the real middle turns AND was fed back
into every subsequent iterative-update prompt, compounding the loss across
compactions.

Guards added at all four summarization sites (whole bug class):
- _generate_summary: length stop raises, gets the existing one-shot
  main-model fallback (a larger output budget may finish the summary), and
  on terminal failure ABORTS compression preserving the session unchanged
  (new _last_summary_truncated_failure flag, same class as empty-content).
- _micro_summarize_one: partial rolling-summary merge is discarded; the
  exchange stays unabsorbed for a later pass.
- _build_chunk_digests: partial lean digest degrades to the
  recover-via-session_search placeholder.
- trajectory_compressor (sync + async): length stop raises into the
  existing retry/backoff loop.

_response_finish_reason() reads dict- and object-shaped responses and
returns "" when the provider omits the field, so proxies that never send
finish_reason are unaffected.

Ported from earendil-works/pi commit 97fa14e39 (pi#7048), adapted to
hermes' abort-preserving compression failure machinery.

Tests: tests/agent/test_compressor_truncated_summary_guard.py (12 tests;
sabotage-verified — disabling the guards fails 4).

agent/context_compressor.py

@@ -166,6 +166,41 @@ def _is_hygiene_idle_timeout_error(error: object) -> bool:
     return any(marker in text for marker in _HYGIENE_IDLE_TIMEOUT_MARKERS)
 
 
+def _response_finish_reason(response: Any) -> str:
+    """Return ``choices[0].finish_reason`` from a dict- or object-shaped response.
+
+    Mirrors the defensive message extraction in ``_generate_summary``: some
+    OpenAI-compatible proxies / local backends return plain dicts, others
+    return SDK objects, and either may omit the field entirely. Returns the
+    lowercased finish reason, or ``""`` when absent/unreadable.
+    """
+    try:
+        if isinstance(response, dict):
+            choices = response.get("choices") or [{}]
+            first = choices[0] if choices else {}
+            reason = (
+                first.get("finish_reason")
+                if isinstance(first, dict)
+                else getattr(first, "finish_reason", None)
+            )
+        else:
+            choices = getattr(response, "choices", None) or []
+            reason = getattr(choices[0], "finish_reason", None) if choices else None
+        return str(reason).strip().lower() if reason else ""
+    except Exception:
+        return ""
+
+
+# RuntimeError marker raised when the summarizer's generation stopped on the
+# output-token cap (``finish_reason == "length"``). A length stop means the
+# summary text is PARTIAL — persisting it as a compaction checkpoint would
+# silently truncate the conversation's memory and feed the cut-off text back
+# into every subsequent iterative-update prompt. The except-branch classifier
+# below keys on this exact substring, so keep raise sites and the classifier
+# in sync. (Ported from earendil-works/pi#7048 / commit 97fa14e39.)
+_TRUNCATED_SUMMARY_MARKER = "finish_reason=length"
+
+
 def _is_summary_access_or_quota_error(exc: Exception) -> bool:
     """Return True for non-retryable summary auth, permission, or quota errors."""
 
@@ -3540,6 +3575,14 @@ class ContextCompressor(ContextEngine):
         # the session unchanged instead of destroying the middle window for a
         # deterministic placeholder (#94448). Independent of abort_on_summary_failure.
         self._last_summary_empty_content_failure: bool = False
+        # Set when summary generation ultimately fails because the summarizer
+        # stopped on its output-token cap (finish_reason == "length") — the
+        # summary text is PARTIAL and must never become a compaction
+        # checkpoint. compress() must ABORT and preserve the session unchanged
+        # exactly like the empty-content class: a truncated checkpoint
+        # silently destroys the compacted middle and compounds across
+        # iterative updates. (Ported from earendil-works/pi#7048.)
+        self._last_summary_truncated_failure: bool = False
         # retrying on the main model, record the failure so gateway /
         # CLI callers can still warn the user even though compression
         # succeeded.  Silent recovery would hide the broken config.
@@ -5296,6 +5339,23 @@ This compaction should PRIORITISE preserving all information related to the focu
                     f"(provider={self.provider or 'auto'} "
                     f"model={self.summary_model or self.model})"
                 )
+            # A finish_reason of "length" means the summarizer hit its output
+            # token cap mid-generation: the text present is PARTIAL. Persisting
+            # a partial summary as the compaction checkpoint silently truncates
+            # the conversation's memory — the cut-off text replaces the real
+            # middle turns AND is fed back into every subsequent iterative
+            # update prompt, compounding the loss across compactions. Treat it
+            # as a failure so it routes through the same main-model fallback +
+            # abort machinery as other degraded responses instead of becoming
+            # a checkpoint. (Ported from earendil-works/pi#7048.)
+            if _response_finish_reason(response) == "length":
+                raise RuntimeError(
+                    "Context compression summary was truncated "
+                    f"({_TRUNCATED_SUMMARY_MARKER}): generation hit the output "
+                    "token cap and the summary is incomplete "
+                    f"(provider={self.provider or 'auto'} "
+                    f"model={self.summary_model or self.model})"
+                )
             # Strip reasoning blocks the summarizer model may have emitted
             # (<think>...</think> etc. from thinking models like MiniMax,
             # DeepSeek, QwQ). Without this the trace is stored in
@@ -5323,6 +5383,7 @@ This compaction should PRIORITISE preserving all information related to the focu
             self._last_summary_auth_failure = False
             self._last_summary_network_failure = False
             self._last_summary_empty_content_failure = False
+            self._last_summary_truncated_failure = False
             return self._with_summary_prefix(summary)
         except Exception as e:
             # ``call_llm`` raises ``RuntimeError`` for two very different cases:
@@ -5396,6 +5457,16 @@ This compaction should PRIORITISE preserving all information related to the focu
                 or "llm returned none response" in _err_str
                 or "llm returned invalid response" in _err_str
             )
+            # Summarizer stopped on its output-token cap (finish_reason ==
+            # "length"): the summary text is partial and must never become a
+            # compaction checkpoint. Same degraded-response handling shape as
+            # empty content — one main-model retry (a larger/unconstrained
+            # model may finish the summary), then ABORT preserving the session
+            # unchanged. (Ported from earendil-works/pi#7048.)
+            _is_truncated_summary = (
+                isinstance(e, RuntimeError)
+                and _TRUNCATED_SUMMARY_MARKER in _err_str
+            )
             # Authentication, permission, and exhausted-quota failures are NOT
             # transient or fixable by retrying the same request. Flag them so
             # compress() preserves the session instead of rotating into a
@@ -5421,13 +5492,15 @@ This compaction should PRIORITISE preserving all information related to the focu
                     e,
                 )
             if (
-                (_is_model_not_found or _is_timeout or _is_json_decode or _is_streaming_closed or _is_empty_content)
+                (_is_model_not_found or _is_timeout or _is_json_decode or _is_streaming_closed or _is_empty_content or _is_truncated_summary)
                 and self.summary_model
                 and self.summary_model != self.model
                 and not getattr(self, "_summary_model_fallen_back", False)
             ):
                 if _is_json_decode:
                     _reason = "returned invalid JSON"
+                elif _is_truncated_summary:
+                    _reason = "returned a truncated summary (output token cap)"
                 elif _is_empty_content:
                     _reason = "returned empty content"
                 elif _is_model_not_found:
@@ -5487,7 +5560,7 @@ This compaction should PRIORITISE preserving all information related to the focu
                     min(self._consecutive_timeout_failures,
                         len(_TIMEOUT_COOLDOWN_LADDER)) - 1
                 ]
-            elif _is_json_decode or _is_streaming_closed or _is_empty_content:
+            elif _is_json_decode or _is_streaming_closed or _is_empty_content or _is_truncated_summary:
                 _transient_cooldown = 30
             else:
                 _transient_cooldown = 60
@@ -5506,6 +5579,8 @@ This compaction should PRIORITISE preserving all information related to the focu
             # auth-failure carve-out; independent of abort_on_summary_failure.
             if _is_streaming_closed:
                 self._last_summary_network_failure = True
+            elif _is_truncated_summary:
+                self._last_summary_truncated_failure = True
             elif _is_empty_content:
                 self._last_summary_empty_content_failure = True
             logger.warning(
@@ -7003,6 +7078,18 @@ This compaction should PRIORITISE preserving all information related to the focu
             logger.info("micro-summarization call failed: %s", exc)
             return None
 
+        # A length stop means the merged rolling summary is partial —
+        # persisting it would silently drop the tail of the merge and feed
+        # the cut-off text into every later micro-compact pass. Leave the
+        # exchange unabsorbed instead; a later pass retries it.
+        # (Same class as _generate_summary's guard; pi#7048.)
+        if _response_finish_reason(response) == "length":
+            logger.warning(
+                "micro-summarization output hit the token cap "
+                "(finish_reason=length) — discarding partial summary",
+            )
+            return None
+
         message = response.choices[0].message
         if isinstance(message, dict):
             content = message.get("content")
@@ -7593,7 +7680,8 @@ This compaction should PRIORITISE preserving all information related to the focu
         self._last_compress_refused_would_grow = False
         self._last_compression_made_progress = False
         # NOTE: do NOT reset _last_summary_auth_failure,
-        # _last_summary_network_failure, or _last_summary_empty_content_failure
+        # _last_summary_network_failure, _last_summary_empty_content_failure,
+        # or _last_summary_truncated_failure
         # here.  These flags are set by _generate_summary() on a terminal
         # failure and are already cleared on a successful summary.  Resetting them eagerly defeats the cooldown
         # protection: _generate_summary() returns None from the cooldown
@@ -7950,6 +8038,7 @@ This compaction should PRIORITISE preserving all information related to the focu
             or self._last_summary_auth_failure
             or self._last_summary_network_failure
             or self._last_summary_empty_content_failure
+            or self._last_summary_truncated_failure
         ):
             n_skipped = compress_end - compress_start
             self._last_summary_dropped_count = 0  # nothing actually dropped
@@ -7959,6 +8048,8 @@ This compaction should PRIORITISE preserving all information related to the focu
                 telemetry["failure_class"] = "summary_auth_failure"
             elif self._last_summary_network_failure:
                 telemetry["failure_class"] = "summary_network_failure"
+            elif self._last_summary_truncated_failure:
+                telemetry["failure_class"] = "summary_truncated_failure"
             elif self._last_summary_empty_content_failure:
                 telemetry["failure_class"] = "summary_empty_content_failure"
             else:
@@ -7988,6 +8079,16 @@ This compaction should PRIORITISE preserving all information related to the focu
                         "recovers, or continue the conversation as-is.",
                         n_skipped,
                     )
+                elif self._last_summary_truncated_failure:
+                    logger.warning(
+                        "Summary generation failed (output hit the token cap; "
+                        "summary is incomplete) — aborting compression. "
+                        "%d message(s) preserved unchanged; the session was NOT "
+                        "rotated. A truncated summary would silently lose "
+                        "context: retry with /compress, or raise the "
+                        "summarizer's output budget.",
+                        n_skipped,
+                    )
                 elif self._last_summary_empty_content_failure:
                     logger.warning(
                         "Summary generation failed (LLM returned empty content) — "

agent/conversation_compression.py

@@ -362,6 +362,7 @@ _COMPRESSOR_ATTEMPT_STATE_FIELDS = (
     "_last_summary_auth_failure",
     "_last_summary_network_failure",
     "_last_summary_empty_content_failure",
+    "_last_summary_truncated_failure",
     "_last_aux_model_failure_error",
     "_last_aux_model_failure_model",
     "_summary_model_fallen_back",

tests/agent/test_compressor_truncated_summary_guard.py

@@ -0,0 +1,165 @@
+"""Truncated compaction summaries must never become checkpoints.
+
+Port of earendil-works/pi#7048 (commit 97fa14e39): a summarization response
+whose ``finish_reason == "length"`` contains PARTIAL text — the generation
+stopped on the output-token cap mid-summary. Persisting it as the compaction
+checkpoint silently truncates the conversation's memory and feeds the cut-off
+text back into every subsequent iterative-update prompt.
+
+Covers all three compressor summarization sites:
+  1. ``_generate_summary`` (main batch summary) — length stop raises, falls
+     back to main model once, then ABORTS compression preserving messages.
+  2. ``_micro_summarize_one`` (micro-compact rolling summary) — length stop
+     discards the partial merge (returns None) so the exchange stays
+     unabsorbed.
+(The former third site, ``_build_chunk_digests``, was removed on main by
+#96603 — lean digests now ride the single ``_generate_summary`` request, so
+its guard is covered by site 1.)
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from agent.context_compressor import (
+    ContextCompressor,
+    _response_finish_reason,
+)
+
+
+def _mock_response(content="a perfectly fine summary", finish_reason="stop"):
+    resp = MagicMock()
+    choice = MagicMock()
+    choice.message.content = content
+    choice.finish_reason = finish_reason
+    resp.choices = [choice]
+    return resp
+
+
+def _msgs(n=12):
+    return [
+        {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i} " + "x" * 50}
+        for i in range(n)
+    ]
+
+
+class TestResponseFinishReason:
+    def test_object_shaped(self):
+        assert _response_finish_reason(_mock_response(finish_reason="length")) == "length"
+        assert _response_finish_reason(_mock_response(finish_reason="stop")) == "stop"
+
+    def test_dict_shaped(self):
+        resp = {"choices": [{"message": {"content": "x"}, "finish_reason": "LENGTH"}]}
+        assert _response_finish_reason(resp) == "length"
+
+    def test_missing_field_is_empty(self):
+        assert _response_finish_reason({"choices": [{"message": {"content": "x"}}]}) == ""
+        assert _response_finish_reason({"choices": []}) == ""
+        assert _response_finish_reason(None) == ""
+
+
+class TestGenerateSummaryTruncationGuard:
+    def test_length_stop_is_rejected_and_aborts(self):
+        """A length-stopped summary must not become a checkpoint; with no
+        distinct aux model to fall back from, compression ABORTS and the
+        session is preserved unchanged."""
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+            c = ContextCompressor(
+                model="test", quiet_mode=True,
+                protect_first_n=2, protect_last_n=2,
+                abort_on_summary_failure=False,
+            )
+        msgs = _msgs()
+        with patch(
+            "agent.context_compressor.call_llm",
+            return_value=_mock_response("partial summary that got cut o", "length"),
+        ):
+            result = c.compress(msgs, current_tokens=999999, force=True)
+
+        assert result == msgs
+        assert c._last_summary_truncated_failure is True
+        assert c._last_compress_aborted is True
+        assert c._last_summary_fallback_used is False
+        # The partial text must never be stored for iterative updates.
+        assert c._previous_summary is None or "cut o" not in (c._previous_summary or "")
+
+    def test_length_stop_falls_back_to_main_model_once(self):
+        """With a distinct aux summary model, a length stop retries once on
+        the main model (which may have a larger output budget) and succeeds."""
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+            c = ContextCompressor(
+                model="main-model",
+                summary_model_override="small-aux-model",
+                quiet_mode=True,
+            )
+        truncated = _mock_response("partial...", "length")
+        ok = _mock_response("full summary via main model", "stop")
+        with patch(
+            "agent.context_compressor.call_llm",
+            side_effect=[truncated, ok],
+        ) as mock_call:
+            result = c._generate_summary(_msgs(2))
+
+        assert mock_call.call_count == 2
+        assert result is not None
+        assert "full summary via main model" in result
+        assert c._last_summary_truncated_failure is False
+
+    def test_stop_finish_reason_still_succeeds(self):
+        """Control: a normal stop-terminated summary is accepted unchanged."""
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+            c = ContextCompressor(model="test", quiet_mode=True)
+        with patch(
+            "agent.context_compressor.call_llm",
+            return_value=_mock_response("complete summary", "stop"),
+        ):
+            result = c._generate_summary(_msgs(2))
+        assert result is not None
+        assert "complete summary" in result
+
+    def test_missing_finish_reason_still_succeeds(self):
+        """Providers that omit finish_reason entirely must not be rejected."""
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+            c = ContextCompressor(model="test", quiet_mode=True)
+        resp = {"choices": [{"message": {"content": "complete summary"}}]}
+        with patch("agent.context_compressor.call_llm", return_value=resp):
+            result = c._generate_summary(_msgs(2))
+        assert result is not None
+        assert "complete summary" in result
+
+    def test_successful_summary_clears_truncated_flag(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+            c = ContextCompressor(model="test", quiet_mode=True)
+        c._last_summary_truncated_failure = True
+        c._summary_failure_cooldown_until = 0
+        with patch(
+            "agent.context_compressor.call_llm",
+            return_value=_mock_response("fine", "stop"),
+        ):
+            result = c._generate_summary(_msgs(2))
+        assert result is not None
+        assert c._last_summary_truncated_failure is False
+
+
+class TestMicroSummarizeTruncationGuard:
+    def test_length_stop_discards_partial_merge(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+            c = ContextCompressor(model="test", quiet_mode=True)
+        c._micro_compact_rolling_summary = "existing rolling summary"
+        with patch(
+            "agent.auxiliary_client.call_llm",
+            return_value=_mock_response("partial merge tex", "length"),
+        ):
+            result = c._micro_summarize_one("user: hi\nassistant: hello")
+        assert result is None
+
+    def test_stop_finish_reason_merges(self):
+        with patch("agent.context_compressor.get_model_context_length", return_value=100000):
+            c = ContextCompressor(model="test", quiet_mode=True)
+        c._micro_compact_rolling_summary = "existing"
+        with patch(
+            "agent.auxiliary_client.call_llm",
+            return_value=_mock_response("merged summary", "stop"),
+        ):
+            result = c._micro_summarize_one("user: hi\nassistant: hello")
+        assert result == "merged summary"

trajectory_compressor.py

@@ -56,6 +56,31 @@ _project_env = Path(__file__).parent / ".env"
 load_hermes_dotenv(hermes_home=_hermes_home, project_env=_project_env)
 
 
+def _response_finish_reason(response: Any) -> str:
+    """Return ``choices[0].finish_reason`` from a dict- or object-shaped response.
+
+    Local copy of ``agent.context_compressor._response_finish_reason`` —
+    trajectory_compressor is a standalone CLI tool and deliberately avoids
+    importing the (heavy) context compressor module. Returns the lowercased
+    finish reason, or ``""`` when absent/unreadable.
+    """
+    try:
+        if isinstance(response, dict):
+            choices = response.get("choices") or [{}]
+            first = choices[0] if choices else {}
+            reason = (
+                first.get("finish_reason")
+                if isinstance(first, dict)
+                else getattr(first, "finish_reason", None)
+            )
+        else:
+            choices = getattr(response, "choices", None) or []
+            reason = getattr(choices[0], "finish_reason", None) if choices else None
+        return str(reason).strip().lower() if reason else ""
+    except Exception:
+        return ""
+
+
 def _effective_temperature_for_model(
     model: str,
     requested_temperature: float,
@@ -658,6 +683,16 @@ Write only the summary, starting with "[CONTEXT SUMMARY]:" prefix."""
                         _create_kwargs["temperature"] = summary_temperature
                     response = self.client.chat.completions.create(**_create_kwargs)
                 
+                _fr = _response_finish_reason(response)
+                if _fr == "length":
+                    # Length stop = partial summary; storing it as the turn
+                    # replacement silently truncates the trajectory's memory.
+                    # Raise so the retry/backoff loop treats it as a failure
+                    # (pi#7048 class).
+                    raise RuntimeError(
+                        "trajectory summarization hit the output token cap "
+                        "(finish_reason=length); summary is incomplete"
+                    )
                 summary = self._coerce_summary_content(response.choices[0].message.content)
                 return self._ensure_summary_prefix(summary)
                 
@@ -727,6 +762,13 @@ Write only the summary, starting with "[CONTEXT SUMMARY]:" prefix."""
                         _create_kwargs["temperature"] = summary_temperature
                     response = await self._get_async_client().chat.completions.create(**_create_kwargs)
                 
+                if _response_finish_reason(response) == "length":
+                    # Length stop = partial summary; see sync sibling above
+                    # (pi#7048 class).
+                    raise RuntimeError(
+                        "trajectory summarization hit the output token cap "
+                        "(finish_reason=length); summary is incomplete"
+                    )
                 summary = self._coerce_summary_content(response.choices[0].message.content)
                 return self._ensure_summary_prefix(summary)