fix(context): preserve messages when summary quota is exhausted

agent/context_compressor.py

@@ -35,6 +35,37 @@ from agent.redact import redact_sensitive_text
 
 logger = logging.getLogger(__name__)
 
+
+_SUMMARY_ACCESS_OR_QUOTA_ERROR_SUBSTRINGS: tuple[str, ...] = (
+    "invalid api key",
+    "invalid x-api-key",
+    "unauthorized",
+    "authentication",
+    "insufficient_quota",
+    "quota exceeded",
+    "quota_exceeded",
+    "out of funds",
+    "out of credits",
+    "out of credit",
+    "out of extra usage",
+)
+
+
+def _is_summary_access_or_quota_error(exc: Exception) -> bool:
+    """Return True for non-retryable summary auth, permission, or quota errors."""
+
+    status = getattr(exc, "status_code", None) or getattr(
+        getattr(exc, "response", None), "status_code", None
+    )
+    if status in {401, 402, 403}:
+        return True
+
+    err_text = str(exc).lower()
+    if "api key" in err_text and ("invalid" in err_text or "blocked" in err_text):
+        return True
+    return any(marker in err_text for marker in _SUMMARY_ACCESS_OR_QUOTA_ERROR_SUBSTRINGS)
+
+
 HISTORICAL_TASK_HEADING = "## Historical Task Snapshot"
 HISTORICAL_IN_PROGRESS_HEADING = "## Historical In-Progress State"
 HISTORICAL_PENDING_ASKS_HEADING = "## Historical Pending User Asks"
@@ -2314,24 +2345,15 @@ This compaction should PRIORITISE preserving all information related to the focu
             # back to the main model instead of entering a 60-second cooldown.
             # See issue #18458.
             _is_streaming_closed = _is_connection_error(e)
-            # Authentication / permission failures (401/403) are NOT transient
-            # and NOT fixable by retrying the same request: the credential is
-            # invalid/blocked/expired or the endpoint is wrong (e.g. a prod
-            # token sent to a staging inference URL). Flag them so compress()
-            # aborts and preserves the session instead of rotating into a
+            # 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
             # degraded child with a placeholder summary. We still allow the
             # one-shot fallback to the MAIN model below when the failure came
-            # from a distinct auxiliary summary_model (its dedicated creds may
-            # be the only broken thing); only a failure on the main model — or
-            # a fallback that also auth-fails — makes the abort stick.
-            _is_auth_error = (
-                _status in {401, 403}
-                or "invalid api key" in _err_str
-                or "invalid x-api-key" in _err_str
-                or ("api key" in _err_str and ("invalid" in _err_str or "blocked" in _err_str))
-                or "unauthorized" in _err_str
-                or "authentication" in _err_str
-            )
+            # from a distinct auxiliary summary_model; only a failure on the
+            # main model — or a fallback that also access/quota-fails — makes
+            # the abort stick.
+            _is_auth_error = _is_summary_access_or_quota_error(e)
             if _is_auth_error:
                 self._last_summary_auth_failure = True
             if _is_json_decode and not _is_model_not_found and not _is_timeout:

tests/agent/test_context_compressor.py

@@ -11,10 +11,18 @@ from agent.context_compressor import (
     SUMMARY_PREFIX,
     COMPRESSED_SUMMARY_METADATA_KEY,
     _summarize_tool_result,
+    _is_summary_access_or_quota_error,
 )
 from hermes_state import SessionDB
 
 
+class StubProviderError(Exception):
+    def __init__(self, message, *, status_code=None, response=None):
+        super().__init__(message)
+        self.status_code = status_code
+        self.response = response
+
+
 @pytest.fixture()
 def compressor():
     """Create a ContextCompressor with mocked dependencies."""
@@ -825,12 +833,76 @@ class TestAuthFailureAborts:
         ]
 
     def _auth_err(self, status=401):
-        err = Exception(
+        return StubProviderError(
             f"Error code: {status} - "
-            "{'status': 401, 'message': 'Your API key is invalid, blocked or out of funds.'}"
+            "{'status': 401, 'message': 'Your API key is invalid, blocked or out of funds.'}",
+            status_code=status,
         )
-        err.status_code = status
-        return err
+
+    @pytest.mark.parametrize(
+        "message",
+        [
+            "insufficient_quota",
+            "quota exceeded",
+            "quota_exceeded",
+            "out of funds",
+            "out of credits",
+            "out of credit",
+            "out of extra usage",
+        ],
+    )
+    def test_quota_classifier_accepts_explicit_provider_signals(self, message):
+        assert _is_summary_access_or_quota_error(Exception(message)) is True
+
+    @pytest.mark.parametrize(
+        "message",
+        [
+            "billing portal is temporarily unavailable",
+            "usage limit documentation could not be loaded",
+            "rate limit exceeded; retry later",
+            "request timed out",
+        ],
+    )
+    def test_quota_classifier_rejects_transient_or_ambiguous_messages(self, message):
+        assert _is_summary_access_or_quota_error(Exception(message)) is False
+
+    @pytest.mark.parametrize("status", [401, 402, 403])
+    def test_access_classifier_accepts_non_retryable_http_statuses(self, status):
+        err = StubProviderError(
+            "provider rejected summary request",
+            status_code=status,
+        )
+        assert _is_summary_access_or_quota_error(err) is True
+
+    def test_classifier_reads_response_status_code(self):
+        err = StubProviderError(
+            "provider rejected summary request",
+            response=MagicMock(status_code=402),
+        )
+        assert _is_summary_access_or_quota_error(err) is True
+
+    def test_400_out_of_extra_usage_aborts_instead_of_dropping_context(self):
+        """Quota exhaustion preserves the original messages for a later retry."""
+        err = StubProviderError(
+            "Error code: 400 - {'error': {'message': 'out of extra usage'}}",
+            status_code=400,
+        )
+        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 = self._msgs(12)
+        with patch("agent.context_compressor.call_llm", side_effect=err):
+            result = c.compress(msgs, current_tokens=999999, force=True)
+
+        assert result == msgs
+        assert c._last_summary_auth_failure is True
+        assert c._last_compress_aborted is True
+        assert c._last_summary_fallback_used is False
 
     def test_generate_summary_flags_auth_failure(self):
         with patch("agent.context_compressor.get_model_context_length", return_value=100000):