Python: Fix ChatHistoryTruncationReducer deleting system prompt

python/semantic_kernel/contents/history_reducer/chat_history_reducer_utils.py

@@ -66,6 +66,7 @@ def locate_safe_reduction_index(
     target_count: int,
     threshold_count: int = 0,
     offset_count: int = 0,
+    has_system_message: bool = False,
 ) -> int | None:
     """Identify the index of the first message at or beyond the specified target_count.
 
@@ -83,11 +84,21 @@ def locate_safe_reduction_index(
         threshold_count: The threshold beyond target_count required to trigger reduction.
                          If total messages <= (target_count + threshold_count), no reduction occurs.
         offset_count: Optional number of messages to skip at the start (e.g. existing summary messages).
+        has_system_message: Whether the history contains a system message that will be preserved
+                           separately. When True, the target_count is adjusted to account for the
+                           system message being re-added after reduction.
 
     Returns:
         The index that identifies the starting point for a reduced history that does not orphan
         sensitive content. Returns None if reduction is not needed.
     """
+    # Adjust target_count to account for the system message that will be preserved separately.
+    # This matches the .NET SDK behavior.
+    if has_system_message:
+        target_count -= 1
+        if target_count <= 0:
+            return None  # Cannot reduce further; only system message would remain
+
     total_count = len(history)
     threshold_index = total_count - (threshold_count or 0) - target_count
     if threshold_index <= offset_count:

python/semantic_kernel/contents/history_reducer/chat_history_truncation_reducer.py

@@ -15,9 +15,9 @@
 
 from semantic_kernel.contents.history_reducer.chat_history_reducer import ChatHistoryReducer
 from semantic_kernel.contents.history_reducer.chat_history_reducer_utils import (
-    extract_range,
     locate_safe_reduction_index,
 )
+from semantic_kernel.contents.utils.author_role import AuthorRole
 from semantic_kernel.utils.feature_stage_decorator import experimental
 
 logger = logging.getLogger(__name__)
@@ -45,15 +45,32 @@ async def reduce(self) -> Self | None:
 
         logger.info("Performing chat history truncation check...")
 
-        truncation_index = locate_safe_reduction_index(history, self.target_count, self.threshold_count)
+        # Preserve system/developer messages so they are not lost during truncation.
+        # This matches the .NET SDK behavior where system messages are always retained.
+        system_message = next(
+            (msg for msg in history if msg.role in (AuthorRole.SYSTEM, AuthorRole.DEVELOPER)),
+            None,
+        )
+
+        truncation_index = locate_safe_reduction_index(
+            history,
+            self.target_count,
+            self.threshold_count,
+            has_system_message=system_message is not None,
+        )
         if truncation_index is None:
             logger.info(
                 f"No truncation index found. Target count: {self.target_count}, Threshold: {self.threshold_count}"
             )
             return None
 
         logger.info(f"Truncating history to {truncation_index} messages.")
-        truncated_list = extract_range(history, start=truncation_index)
+        truncated_list = history[truncation_index:]
+
+        # Prepend the system/developer message if it was truncated away
+        if system_message is not None and system_message not in truncated_list:
+            truncated_list = [system_message, *truncated_list]
+
         self.messages = truncated_list
         return self
 

python/tests/unit/contents/test_chat_history_truncation_reducer.py

@@ -83,11 +83,13 @@ async def test_truncation_reducer_truncation(chat_messages):
     reducer = ChatHistoryTruncationReducer(target_count=2)
     reducer.messages = chat_messages
     result = await reducer.reduce()
-    # We expect only 2 messages remain after truncation
+    # We expect 2 messages: system message is preserved + 1 conversation message
+    # (target_count=2 includes the system message, matching .NET SDK behavior)
     assert result is not None
     assert len(result) == 2
-    # They should be the last 4 messages while tool result is not orphaned
-    assert result[0] == chat_messages[-2]
+    # System message should be first, followed by the last conversation message
+    assert result[0] == chat_messages[0]  # System message preserved
+    assert result[0].role == AuthorRole.SYSTEM
     assert result[1] == chat_messages[-1]
 
 
@@ -96,12 +98,89 @@ async def test_truncation_reducer_truncation_with_tools(chat_messages_with_tools
     reducer = ChatHistoryTruncationReducer(target_count=3, threshold_count=0)
     reducer.messages = chat_messages_with_tools
     result = await reducer.reduce()
-    # We expect 4 messages remain after truncation
-    # Tool results are not orphaned, so we expect to keep them
+    # We expect 3 messages: system message + last 2 conversation messages
+    # (target_count=3 includes the system message, matching .NET SDK behavior)
     assert result is not None
-    assert len(result) == 4
-    # They should be the last 4 messages
-    assert result[0] == chat_messages_with_tools[-4]
-    assert result[1] == chat_messages_with_tools[-3]
-    assert result[2] == chat_messages_with_tools[-2]
-    assert result[3] == chat_messages_with_tools[-1]
+    assert len(result) == 3
+    # System message preserved, followed by last user-assistant pair
+    assert result[0] == chat_messages_with_tools[0]  # System message
+    assert result[0].role == AuthorRole.SYSTEM
+    assert result[1] == chat_messages_with_tools[-2]  # User message 2
+    assert result[2] == chat_messages_with_tools[-1]  # Assistant message 2
+
+
+async def test_truncation_preserves_system_message():
+    """Verify that the system message is preserved after truncation (issue #12612)."""
+    reducer = ChatHistoryTruncationReducer(
+        target_count=2,
+        system_message="a system message",
+    )
+    reducer.add_message(ChatMessageContent(role=AuthorRole.USER, content="user prompt 1"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.ASSISTANT, content="response 1"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.USER, content="user prompt 2"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.ASSISTANT, content="response 2"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.USER, content="user prompt 3"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.ASSISTANT, content="response 3"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.USER, content="user prompt 4"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.ASSISTANT, content="response 4"))
+
+    result = await reducer.reduce()
+    assert result is not None
+
+    # System message must be present after reduction
+    roles = [msg.role for msg in result.messages]
+    assert AuthorRole.SYSTEM in roles, "System message was deleted by the history reducer"
+    assert result.messages[0].role == AuthorRole.SYSTEM
+    assert result.messages[0].content == "a system message"
+
+
+async def test_truncation_preserves_developer_message():
+    """Verify that developer messages are preserved after truncation."""
+    msgs = [
+        ChatMessageContent(role=AuthorRole.DEVELOPER, content="Developer instructions."),
+        ChatMessageContent(role=AuthorRole.USER, content="User message 1"),
+        ChatMessageContent(role=AuthorRole.ASSISTANT, content="Assistant message 1"),
+        ChatMessageContent(role=AuthorRole.USER, content="User message 2"),
+        ChatMessageContent(role=AuthorRole.ASSISTANT, content="Assistant message 2"),
+    ]
+    reducer = ChatHistoryTruncationReducer(target_count=2)
+    reducer.messages = msgs
+    result = await reducer.reduce()
+    assert result is not None
+    assert len(result) == 2
+    # Developer message should be first
+    assert result[0].role == AuthorRole.DEVELOPER
+    assert result[0].content == "Developer instructions."
+
+
+async def test_truncation_target_count_1_with_system_message():
+    """Verify target_count=1 with system message does not crash (edge case from review)."""
+    reducer = ChatHistoryTruncationReducer(target_count=1, system_message="System prompt")
+    reducer.add_message(ChatMessageContent(role=AuthorRole.USER, content="Hello"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.ASSISTANT, content="Hi"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.USER, content="How are you?"))
+    reducer.add_message(ChatMessageContent(role=AuthorRole.ASSISTANT, content="Good"))
+
+    # Should not crash. Either returns None (no reduction possible)
+    # or returns just the system message.
+    result = await reducer.reduce()
+    if result is not None:
+        # System message must be preserved
+        assert any(m.role == AuthorRole.SYSTEM for m in result.messages)
+
+
+async def test_truncation_without_system_message():
+    """Verify truncation works correctly when there is no system message."""
+    msgs = [
+        ChatMessageContent(role=AuthorRole.USER, content="User message 1"),
+        ChatMessageContent(role=AuthorRole.ASSISTANT, content="Assistant message 1"),
+        ChatMessageContent(role=AuthorRole.USER, content="User message 2"),
+        ChatMessageContent(role=AuthorRole.ASSISTANT, content="Assistant message 2"),
+    ]
+    reducer = ChatHistoryTruncationReducer(target_count=2)
+    reducer.messages = msgs
+    result = await reducer.reduce()
+    assert result is not None
+    assert len(result) == 2
+    assert result[0] == msgs[-2]
+    assert result[1] == msgs[-1]