fix: preserve false bypass values and single hook invocation

agent/context_compressor_summary.py

@@ -37,8 +37,8 @@ class SummaryDispatchMixin:
             "focus_topic": focus_topic or self._derive_auto_focus_topic(messages),
             "memory_context": memory_context,
         }
-        if bypass_cooldown and _accepts_keyword_argument(self._generate_summary, "bypass_cooldown"):
-            summary_kwargs["bypass_cooldown"] = True
+        if _accepts_keyword_argument(self._generate_summary, "bypass_cooldown"):
+            summary_kwargs["bypass_cooldown"] = bypass_cooldown
         try:
             return self._generate_summary(turns_to_summarize, **summary_kwargs)
         except AuxiliaryExplicitCancellation:

tests/agent/test_context_compressor_summary_hook_compat.py

@@ -1,5 +1,7 @@
 """Compatibility coverage for third-party summary hook overrides."""
 
+import pytest
+
 from agent.context_compressor import ContextCompressor
 
 
@@ -8,71 +10,66 @@ class _LegacySummaryCompressor(ContextCompressor):
 
     def __init__(self) -> None:
         super().__init__(
-            model="test-model",
-            protect_first_n=2,
-            protect_last_n=2,
-            quiet_mode=True,
-            config_context_length=40_960,
+            model="test-model", protect_first_n=2, protect_last_n=2,
+            quiet_mode=True, config_context_length=40_960,
         )
-        self.summary_calls: list[tuple[list[dict], str | None, str]] = []
+        self.summary_calls = []
 
-    def _generate_summary(
-        self,
-        turns_to_summarize: list[dict],
-        focus_topic: str | None = None,
-        memory_context: str = "",
-    ) -> str:
-        self.summary_calls.append((turns_to_summarize, focus_topic, memory_context))
+    def _generate_summary(self, turns_to_summarize, focus_topic=None, memory_context=""):
+        self.summary_calls.append((focus_topic, memory_context, {}))
         return "## Goal\nPreserve compatibility with legacy summary hooks."
 
 
 class _KwargsSummaryCompressor(_LegacySummaryCompressor):
-    def __init__(self) -> None:
-        super().__init__()
-        self.summary_kwargs: list[dict[str, object]] = []
-
-    def _generate_summary(
-        self,
-        turns_to_summarize: list[dict],
-        focus_topic: str | None = None,
-        memory_context: str = "",
-        **kwargs: object,
-    ) -> str:
-        self.summary_kwargs.append(kwargs)
+    def _generate_summary(self, turns_to_summarize, focus_topic=None, memory_context="", **kwargs):
+        self.summary_calls.append((focus_topic, memory_context, kwargs))
         return "## Goal\nPreserve bypass semantics for extensible hooks."
 
 
-def _messages() -> list[dict]:
-    messages = [{"role": "system", "content": "system"}]
-    messages.extend(
-        {
-            "role": "user" if index % 2 == 0 else "assistant",
-            "content": f"turn-{index} " + "context " * 1_000,
-        }
-        for index in range(14)
-    )
-    return messages
+class _ExplicitSummaryCompressor(_KwargsSummaryCompressor):
+    def _generate_summary(self, turns_to_summarize, focus_topic=None, memory_context="", *, bypass_cooldown=True):
+        return super()._generate_summary(
+            turns_to_summarize, focus_topic, memory_context, bypass_cooldown=bypass_cooldown,
+        )
 
 
-def test_compress_omits_bypass_cooldown_for_legacy_summary_override() -> None:
-    compressor = _LegacySummaryCompressor()
+class _PositionalSummaryCompressor(_LegacySummaryCompressor):
+    def _generate_summary(self, turns_to_summarize, bypass_cooldown=False, /, focus_topic=None, memory_context=""):
+        return super()._generate_summary(turns_to_summarize, focus_topic, memory_context)
+
+
+def _messages():
+    return [{"role": "system", "content": "system"}] + [
+        {"role": "user" if i % 2 == 0 else "assistant", "content": f"turn-{i} " + "context " * 1_000}
+        for i in range(14)
+    ]
+
+
+@pytest.mark.parametrize("bypass", [False, True])
+@pytest.mark.parametrize("engine", [
+    _LegacySummaryCompressor, _KwargsSummaryCompressor,
+    _ExplicitSummaryCompressor, _PositionalSummaryCompressor,
+])
+def test_compress_calls_supported_summary_signature_once(engine, bypass):
+    compressor = engine()
     messages = _messages()
-
     compressed = compressor.compress(
-        messages,
-        current_tokens=30_000,
-        memory_context="plugin memory",
-        bypass_cooldown=True,
+        messages, current_tokens=30_000, focus_topic="compatibility",
+        memory_context="plugin memory", bypass_cooldown=bypass,
     )
-
+    expected = {"bypass_cooldown": bypass} if isinstance(compressor, _KwargsSummaryCompressor) else {}
     assert len(compressed) < len(messages)
+    assert compressor.summary_calls == [("compatibility", "plugin memory", expected)]
+
+
+@pytest.mark.parametrize("bypass", [False, True])
+def test_summary_hook_type_error_is_not_retried(bypass):
+    class FailingCompressor(_KwargsSummaryCompressor):
+        def _generate_summary(self, *args, **kwargs):
+            super()._generate_summary(*args, **kwargs)
+            raise TypeError("inside stateful hook")
+
+    compressor = FailingCompressor()
+    with pytest.raises(TypeError, match="inside stateful hook"):
+        compressor.compress(_messages(), current_tokens=30_000, bypass_cooldown=bypass)
     assert len(compressor.summary_calls) == 1
-    assert compressor.summary_calls[0][2] == "plugin memory"
-
-
-def test_compress_passes_bypass_cooldown_to_kwargs_summary_override() -> None:
-    compressor = _KwargsSummaryCompressor()
-
-    compressor.compress(_messages(), current_tokens=30_000, bypass_cooldown=True)
-
-    assert compressor.summary_kwargs == [{"bypass_cooldown": True}]