diff --git a/agent/context_compressor.py b/agent/context_compressor.py index b443dfef5d..5c613b8618 100644 --- a/agent/context_compressor.py +++ b/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 # (... 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) — " diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index fa9ea29a04..19056ad185 100644 --- a/agent/conversation_compression.py +++ b/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", diff --git a/tests/agent/test_compressor_truncated_summary_guard.py b/tests/agent/test_compressor_truncated_summary_guard.py new file mode 100644 index 0000000000..edb326cef0 --- /dev/null +++ b/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" diff --git a/trajectory_compressor.py b/trajectory_compressor.py index 8feb4a5157..58c0ca3245 100644 --- a/trajectory_compressor.py +++ b/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)