Python: fix: preserve synthetic compaction summaries

python/packages/core/agent_framework/_compaction.py

@@ -1579,15 +1579,34 @@ async def before_run(
         if not all_messages:
             return
 
+        # Track each original message's source before compaction
+        source_by_id: dict[int, str] = {
+            id(message): sid for sid, msgs in context.context_messages.items() for message in msgs
+        }
+
         annotate_message_groups(all_messages)
         if self.tokenizer is not None:
             annotate_token_counts(all_messages, tokenizer=self.tokenizer)
         await self.before_strategy(all_messages)
 
         projected = project_included_messages(all_messages)
-        projected_set = {id(m) for m in projected}
-        for sid in list(context.context_messages):
-            context.context_messages[sid] = [m for m in context.context_messages[sid] if id(m) in projected_set]
+
+        # Rebuild provider message lists from the projected list, preserving source attribution
+        # and including new synthetic messages created by compaction strategies
+        rebuilt: dict[str, list[Message]] = {sid: [] for sid in context.context_messages}
+        fallback_sid = next(iter(rebuilt), self.source_id)
+        last_sid = fallback_sid
+        for message in projected:
+            # For new synthetic messages, use the last known source; for original messages, use their tracked source
+            sid = source_by_id.get(id(message), last_sid)
+            if sid not in rebuilt:
+                # If the source was somehow removed during compaction, fall back to the last known source
+                sid = last_sid
+            rebuilt[sid].append(message)
+            last_sid = sid
+
+        context.context_messages.clear()
+        context.context_messages.update(rebuilt)
 
     async def after_run(
         self,

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

@@ -1939,3 +1939,125 @@ def test_serialize_message_preserves_non_ascii_for_token_count() -> None:
     assert text in serialized
     assert "\\u3053" not in serialized
     assert tokenizer.count_tokens(serialized) < tokenizer.count_tokens(escaped)
+
+
+async def test_compaction_provider_before_run_preserves_synthetic_summary_messages() -> None:
+    """Test that before_run preserves synthetic summary messages created by compaction strategies.
+
+    This is a regression test for a bug where ToolResultCompactionStrategy and SummarizationStrategy
+    create new synthetic Message objects during compaction, but CompactionProvider.before_run only
+    filtered existing messages by id(), causing the new summary messages to be silently dropped.
+    """
+    from agent_framework._sessions import SessionContext
+
+    # Create a session context with some messages from a history provider
+    messages = [
+        Message(role="user", contents=["hello"]),
+        _assistant_function_call("c1"),
+        _tool_result("c1", "sunny, 18C"),
+        Message(role="assistant", contents=["final response"]),
+    ]
+
+    ctx = SessionContext(input_messages=[])
+    ctx.extend_messages("history", messages)
+
+    # Verify initial state - 4 messages from history provider
+    assert len(ctx.context_messages["history"]) == 4
+    assert len(ctx.get_messages()) == 4
+
+    # Apply ToolResultCompactionStrategy via CompactionProvider.before_run
+    provider = CompactionProvider(
+        before_strategy=ToolResultCompactionStrategy(keep_last_tool_call_groups=0)
+    )
+
+    await provider.before_run(agent=None, session=None, context=ctx, state={})
+
+    # After compaction, we should have 3 messages:
+    # - user message
+    # - synthetic summary message "[Tool results: tool: sunny, 18C]"
+    # - final assistant response
+    final_messages = ctx.get_messages()
+    assert len(final_messages) == 3, f"Expected 3 messages after compaction, got {len(final_messages)}"
+
+    # Verify the summary message is present - it should be an assistant message with tool results summary
+    summary_messages = [
+        m for m in final_messages 
+        if m.role == "assistant" and any(hasattr(c, 'text') and "Tool results" in c.text for c in m.contents)
+    ]
+    assert len(summary_messages) == 1, "Summary message should be present in final messages"
+
+    # Verify the context_messages dict was also updated correctly
+    assert len(ctx.context_messages["history"]) == 3, "History provider should have 3 messages after compaction"
+
+    # Verify the summary is in the history provider's list
+    history_summary_messages = [
+        m for m in ctx.context_messages["history"]
+        if m.role == "assistant" and any(hasattr(c, 'text') and "Tool results" in c.text for c in m.contents)
+    ]
+    assert len(history_summary_messages) == 1, "Summary message should be in history provider's message list"
+
+
+async def test_compaction_provider_before_run_preserves_summarization_strategy_messages() -> None:
+    """Test that before_run preserves synthetic summary messages created by SummarizationStrategy.
+
+    This is a regression test for the same bug affecting SummarizationStrategy, which creates
+    new synthetic Message objects during compaction.
+    """
+    from agent_framework._sessions import SessionContext
+    from unittest.mock import AsyncMock, MagicMock
+
+    # Create a mock chat client for summarization
+    mock_client = AsyncMock()
+    mock_response = MagicMock()
+    mock_response.text = "Summary of conversation"
+    mock_client.get_response.return_value = mock_response
+
+    # Create messages that will trigger summarization
+    messages = [
+        Message(role="user", contents=["first message"]),
+        Message(role="assistant", contents=["first response"]),
+        Message(role="user", contents=["second message"]),
+        Message(role="assistant", contents=["second response"]),
+        Message(role="user", contents=["third message"]),
+        Message(role="assistant", contents=["third response"]),
+    ]
+
+    ctx = SessionContext(input_messages=[])
+    ctx.extend_messages("history", messages)
+
+    # Verify initial state - 6 messages from history provider
+    assert len(ctx.context_messages["history"]) == 6
+    assert len(ctx.get_messages()) == 6
+
+    # Apply SummarizationStrategy via CompactionProvider.before_run
+    # Set target_count=2 to trigger summarization since we have 6 non-system messages
+    provider = CompactionProvider(
+        before_strategy=SummarizationStrategy(
+            client=mock_client,
+            target_count=2,
+            threshold=0,
+        )
+    )
+
+    await provider.before_run(agent=None, session=None, context=ctx, state={})
+
+    # After compaction, we should have messages including the synthetic summary
+    final_messages = ctx.get_messages()
+    assert len(final_messages) >= 3, f"Expected at least 3 messages after compaction (summary + retained), got {len(final_messages)}"
+
+    # Verify the summary message is present
+    summary_messages = [
+        m for m in final_messages 
+        if m.role == "assistant" and any(hasattr(c, 'text') and "Summary of conversation" in c.text for c in m.contents)
+    ]
+    assert len(summary_messages) == 1, "Summary message should be present in final messages"
+
+    # Verify the context_messages dict was also updated correctly
+    assert len(ctx.context_messages["history"]) >= 3, "History provider should have at least 3 messages after compaction"
+
+    # Verify the summary is in the history provider's list
+    history_summary_messages = [
+        m for m in ctx.context_messages["history"]
+        if m.role == "assistant" and any(hasattr(c, 'text') and "Summary of conversation" in c.text for c in m.contents)
+    ]
+    assert len(history_summary_messages) == 1, "Summary message should be in history provider's message list"