Python: Bound tool result compaction summaries

python/packages/core/agent_framework/_compaction.py

@@ -980,6 +980,11 @@ class ToolResultCompactionStrategy:
     untouched; older ones are collapsed.
     """
 
+    _SUMMARY_PREFIX = "[Tool results: "
+    _SUMMARY_SUFFIX = "]"
+    _SUMMARY_MAX_CHARS = 4096
+    _SUMMARY_TRUNCATION_MARKER = "... [truncated]"
+
     def __init__(self, *, keep_last_tool_call_groups: int = 1) -> None:
         """Create a tool-result compaction strategy.
 
@@ -1019,30 +1024,27 @@ async def __call__(self, messages: list[Message]) -> bool:
             if group_id in keep_ids:
                 continue
             group_msgs = grouped.get(group_id, [])
-            # Build a call_id → function_name map from function_call contents.
             call_id_to_name: dict[str, str] = {}
+            tool_results: list[str] = []
             for msg in group_msgs:
+                if msg.additional_properties.get(EXCLUDED_KEY, False):
+                    continue
                 for content in msg.contents:
                     if content.type == "function_call" and content.call_id and content.name:
                         call_id_to_name[content.call_id] = content.name
                     elif content.type == "mcp_server_tool_call" and content.call_id and content.tool_name:
                         call_id_to_name[content.call_id] = content.tool_name
-            # Collect tool results with the function name for context.
-            tool_results: list[str] = []
-            for msg in group_msgs:
-                for content in msg.contents:
-                    if content.type == "function_result":
+                    elif content.type == "function_result":
                         result_text = content.result if isinstance(content.result, str) else str(content.result)
-                        func_name = call_id_to_name.get(content.call_id or "", "")
-                        label = f"{func_name}: {result_text}" if func_name else result_text
+                        tool_name = call_id_to_name.get(content.call_id or "", "")
+                        label = f"{tool_name}: {result_text}" if tool_name else result_text
                         tool_results.append(label.strip())
                     elif content.type == "mcp_server_tool_result":
                         result_text = _tool_result_text(content.output)
                         tool_name = call_id_to_name.get(content.call_id or "", "")
                         label = f"{tool_name}: {result_text}" if tool_name else result_text
                         tool_results.append(label.strip())
-            summary_label = "; ".join(tool_results) if tool_results else "no results"
-            summary_text = f"[Tool results: {summary_label}]"
+            summary_text = self._summary_text(tool_results)
 
             summary_id = f"tool_summary_{group_id}"
             original_message_ids = [msg.message_id for msg in group_msgs if msg.message_id]
@@ -1073,6 +1075,24 @@ async def __call__(self, messages: list[Message]) -> bool:
 
         return changed
 
+    @classmethod
+    def _summary_text(cls, tool_results: list[str]) -> str:
+        summary_label = "; ".join(tool_results) if tool_results else "no results"
+        summary_text = f"{cls._SUMMARY_PREFIX}{summary_label}{cls._SUMMARY_SUFFIX}"
+        if len(summary_text) <= cls._SUMMARY_MAX_CHARS:
+            return summary_text
+
+        allowed_label_chars = (
+            cls._SUMMARY_MAX_CHARS
+            - len(cls._SUMMARY_PREFIX)
+            - len(cls._SUMMARY_TRUNCATION_MARKER)
+            - len(cls._SUMMARY_SUFFIX)
+        )
+        if allowed_label_chars <= 0:
+            return f"{cls._SUMMARY_PREFIX}{cls._SUMMARY_TRUNCATION_MARKER}{cls._SUMMARY_SUFFIX}"
+        truncated_label = summary_label[:allowed_label_chars].rstrip()
+        return f"{cls._SUMMARY_PREFIX}{truncated_label}{cls._SUMMARY_TRUNCATION_MARKER}{cls._SUMMARY_SUFFIX}"
+
 
 def _tool_result_text(value: Any) -> str:
     if isinstance(value, str):

python/packages/core/tests/core/test_compaction.py

@@ -1287,6 +1287,64 @@ async def test_tool_result_compaction_preserves_tool_results_in_summary() -> Non
     assert "found 3 docs" in summary_msgs[0].text  # type: ignore[operator]
 
 
+async def test_tool_result_compaction_bounds_large_summary_payload() -> None:
+    """Summary text should not embed an oversized tool result verbatim."""
+    payload_line = "line contents\n"
+    payload_lines = ToolResultCompactionStrategy._SUMMARY_MAX_CHARS // len(payload_line) + 1
+    large_result = "file-start\n" + (payload_line * payload_lines) + "file-end"
+    messages = [
+        Message(role="user", contents=["read the file"]),
+        Message(
+            role="assistant",
+            contents=[Content.from_function_call(call_id="c1", name="read_file", arguments="{}")],
+        ),
+        _tool_result("c1", large_result),
+        Message(role="assistant", contents=["done"]),
+    ]
+    strategy = ToolResultCompactionStrategy(keep_last_tool_call_groups=0)
+    annotate_message_groups(messages)
+
+    await strategy(messages)
+
+    summary = next(m for m in included_messages(messages) if (m.text or "").startswith("[Tool results:"))
+    summary_text = summary.text or ""
+    assert large_result not in summary_text
+    assert "file-start" in summary_text
+    assert "file-end" not in summary_text
+    assert "[truncated]" in summary_text
+    assert len(summary_text) <= ToolResultCompactionStrategy._SUMMARY_MAX_CHARS
+    assert len(summary_text) < len(large_result)
+
+
+async def test_tool_result_compaction_ignores_already_excluded_results() -> None:
+    """Summary text should not restore tool results already excluded from context."""
+    excluded_result = "excluded-start\n" + ("excluded line\n" * 1_000) + "excluded-end"
+    messages = [
+        Message(role="user", contents=["read the files"]),
+        Message(
+            role="assistant",
+            contents=[
+                Content.from_function_call(call_id="c1", name="read_file", arguments="{}"),
+                Content.from_function_call(call_id="c2", name="read_file", arguments="{}"),
+            ],
+        ),
+        _tool_result("c1", "included result"),
+        _tool_result("c2", excluded_result),
+        Message(role="assistant", contents=["done"]),
+    ]
+    strategy = ToolResultCompactionStrategy(keep_last_tool_call_groups=0)
+    annotate_message_groups(messages)
+    messages[3].additional_properties[EXCLUDED_KEY] = True
+
+    await strategy(messages)
+
+    summary = next(m for m in included_messages(messages) if (m.text or "").startswith("[Tool results:"))
+    summary_text = summary.text or ""
+    assert "included result" in summary_text
+    assert excluded_result not in summary_text
+    assert "excluded-start" not in summary_text
+
+
 async def test_tool_result_compaction_bidirectional_tracing() -> None:
     """Summary and originals should link to each other like SummarizationStrategy does."""
     messages = [