fix: [BUG] SummarizingConversationManager's pin_first does not work correctly with FileSessionManager and S3SessionManager (#4027) [replica of harness-sdk PR 4066]

strands-py/src/strands/agent/conversation_manager/compression/pin_message.py

@@ -55,6 +55,24 @@ def is_pinned(messages: Messages, index: int) -> bool:
     return False
 
 
+def count_protected_prefix(messages: Messages) -> int:
+    """Count the leading messages protected from eviction.
+
+    ``is_pinned`` protects a pinned message's adjacent tool-pair partner as well, so the protected
+    head can extend past the explicitly pinned range.
+
+    Args:
+        messages: The full messages array.
+
+    Returns:
+        The number of leading messages that ``is_pinned`` protects.
+    """
+    count = 0
+    while count < len(messages) and is_pinned(messages, count):
+        count += 1
+    return count
+
+
 def apply_pin_first(messages: Messages, count: int) -> None:
     """Pin the first N messages in the array permanently.
 

strands-py/src/strands/agent/conversation_manager/conversation_manager.py

@@ -78,6 +78,9 @@ def __init__(self, *, proactive_compression: Union[bool, "ProactiveCompressionCo
           removed_message_count: The messages that have been removed from the agents messages array.
               These represent messages provided by the user or LLM that have been removed, not messages
               included by the conversation manager through something like summarization.
+          pinned_head_count: The number of leading messages the manager has protected from eviction.
+              Restoring a session reattaches this many stored messages, so it counts every protected
+              message, including the tool-pair partner of a pinned message.
         """
         # Resolve the threshold from proactive_compression parameter
         if proactive_compression is True:
@@ -91,6 +94,7 @@ def __init__(self, *, proactive_compression: Union[bool, "ProactiveCompressionCo
             raise ValueError(f"compression_threshold must be between 0 (exclusive) and 1 (inclusive), got {threshold}")
 
         self.removed_message_count = 0
+        self.pinned_head_count = 0
         self._compression_threshold = threshold
 
     def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
@@ -154,13 +158,15 @@ def restore_from_session(self, state: dict[str, Any]) -> list[Message] | None:
         if state.get("__name__") != self.__class__.__name__:
             raise ValueError("Invalid conversation manager state.")
         self.removed_message_count = state["removed_message_count"]
+        self.pinned_head_count = state.get("pinned_head_count", 0)
         return None
 
     def get_state(self) -> dict[str, Any]:
         """Get the current state of a Conversation Manager as a Json serializable dictionary."""
         return {
             "__name__": self.__class__.__name__,
             "removed_message_count": self.removed_message_count,
+            "pinned_head_count": self.pinned_head_count,
         }
 
     @abstractmethod

strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

@@ -11,7 +11,7 @@
 from ...types.exceptions import ContextWindowOverflowException
 from ...types.tools import ToolResultContent
 from .compression.context_compression import find_valid_trim_point
-from .compression.pin_message import apply_pin_first, is_pinned
+from .compression.pin_message import apply_pin_first, count_protected_prefix, is_pinned
 from .conversation_manager import ConversationManager, ProactiveCompressionConfig
 
 logger = logging.getLogger(__name__)
@@ -79,7 +79,6 @@ def __init__(
         self.should_truncate_results = should_truncate_results
         self.per_turn = per_turn
         self.pin_first = max(0, pin_first) if pin_first is not None else None
-        self._pin_first_applied = False
         self._model_call_count = 0
 
     def register_hooks(self, registry: "HookRegistry", **kwargs: Any) -> None:
@@ -196,9 +195,11 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
         messages = agent.messages
 
         # Pin first N messages permanently (only on first reduction)
-        if self.pin_first and not self._pin_first_applied:
+        if self.pin_first and self.pinned_head_count == 0:
             apply_pin_first(messages, self.pin_first)
-            self._pin_first_applied = True
+            # is_pinned also protects a pinned message's tool-pair partner, so the protected head can
+            # be longer than pin_first; restore reattaches exactly this many stored messages.
+            self.pinned_head_count = count_protected_prefix(messages)
 
         # window_size=0 means "remove all non-pinned messages"
         if self.window_size == 0:

strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py

@@ -79,7 +79,6 @@ def __init__(
         self.summarization_agent = summarization_agent
         self.summarization_system_prompt = summarization_system_prompt
         self.pin_first = max(0, pin_first) if pin_first is not None else None
-        self._pin_first_applied = False
         self._summary_message: Message | None = None
 
     @override
@@ -98,7 +97,10 @@ def restore_from_session(self, state: dict[str, Any]) -> list[Message] | None:
 
     def get_state(self) -> dict[str, Any]:
         """Returns a dictionary representation of the state for the Summarizing Conversation Manager."""
-        return {"summary_message": self._summary_message, **super().get_state()}
+        return {
+            "summary_message": self._summary_message,
+            **super().get_state(),
+        }
 
     def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         """Apply management strategy to conversation history.
@@ -147,6 +149,9 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
     def _summarize_oldest(self, agent: "Agent") -> None:
         """Summarize the oldest messages and replace them with a summary.
 
+        Bookkeeping updates (removed_message_count, pinned_head_count) are applied only after
+        summary generation succeeds, making the operation atomic by construction.
+
         Args:
             agent: The agent instance.
 
@@ -173,9 +178,8 @@ def _summarize_oldest(self, agent: "Agent") -> None:
             raise ContextWindowOverflowException("Cannot summarize: insufficient messages for summarization")
 
         # Pin first N messages permanently (only on first reduction)
-        if self.pin_first and not self._pin_first_applied:
+        if self.pin_first and self.pinned_head_count == 0:
             apply_pin_first(agent.messages, self.pin_first)
-            self._pin_first_applied = True
 
         # Partition [0, messages_to_summarize_count) into pinned (preserve) and non-pinned (summarize)
         protected_to_preserve, to_summarize = partition_pinned(agent.messages, 0, messages_to_summarize_count)
@@ -185,16 +189,17 @@ def _summarize_oldest(self, agent: "Agent") -> None:
 
         remaining_messages = agent.messages[messages_to_summarize_count:]
 
-        # Keep track of the number of messages that have been summarized thus far.
-        self.removed_message_count += len(to_summarize)
-        # If there is a summary message, don't count it in the removed_message_count.
-        if self._summary_message:
-            self.removed_message_count -= 1
+        # Generate summary — if this fails, no bookkeeping is updated
+        summary_message = self._generate_summary(to_summarize, agent)
+        _ensure_tracking_id(summary_message)
 
-        # Generate summary
-        self._summary_message = self._generate_summary(to_summarize, agent)
-        # Assign tracking id to the summary message since it bypasses the append method.
-        _ensure_tracking_id(self._summary_message)
+        # All mutations below this point: summary generation succeeded, commit the changes.
+        removed_count = len(to_summarize)
+        if self._summary_message:
+            removed_count -= 1
+        self.removed_message_count += removed_count
+        self.pinned_head_count = len(protected_to_preserve)
+        self._summary_message = summary_message
 
         # Replace summarized range with protected messages + summary + remaining
         agent.messages[:] = protected_to_preserve + [self._summary_message] + remaining_messages

strands-py/src/strands/session/repository_session_manager.py

@@ -4,6 +4,7 @@
 import logging
 from typing import TYPE_CHECKING, Any
 
+from ..agent.conversation_manager.compression.pin_message import pin_message
 from ..agent.state import AgentState
 from ..tools._tool_helpers import generate_missing_tool_result_content
 from ..types.agent import LocalAgent
@@ -263,22 +264,49 @@ def initialize(self, agent: "LocalAgent", **kwargs: Any) -> None:
             # Restore the conversation manager to its previous state, and get the optional prepend messages
             prepend_messages = []
             offset = 0
+            pinned_head_count = 0
             if isinstance(agent, Agent):
                 prepend_messages = (
                     agent.conversation_manager.restore_from_session(session_agent.conversation_manager_state) or []
                 )
                 offset = agent.conversation_manager.removed_message_count
+                # The pinned head count is persisted in the base ConversationManager and restored above.
+                pinned_head_count = agent.conversation_manager.pinned_head_count
+
+            pinned_messages: list[Message] = []
+            if pinned_head_count > 0:
+                pinned_session_messages = self.session_repository.list_messages(
+                    session_id=self.session_id,
+                    agent_id=agent.agent_id,
+                    limit=pinned_head_count,
+                    offset=0,
+                )
+                pinned_messages = [session_message.to_message() for session_message in pinned_session_messages]
+                # Re-apply pin markers to the restored head so the next compaction respects them.
+                for i in range(len(pinned_messages)):
+                    pin_message(pinned_messages, i)
 
             # List the messages currently in the session, using an offset of the messages previously removed
-            # by the conversation manager.
+            # by the conversation manager. The offset must also skip past the still-live pinned head so the
+            # tail does not overlap with the messages reattached as ``pinned_messages`` above.
             session_messages = self.session_repository.list_messages(
                 session_id=self.session_id,
                 agent_id=agent.agent_id,
-                offset=offset,
+                offset=pinned_head_count + offset,
             )
             self._held_records[agent.agent_id] = list(session_messages)
+
+            # When the tail comes back empty but the store is not, seed the append cursor from the
+            # last stored message so appending never restarts at 0 on a non-empty store.
             if len(session_messages) > 0:
                 self._latest_agent_message[agent.agent_id] = session_messages[-1]
+            elif pinned_head_count > 0 or offset > 0:
+                all_messages = self.session_repository.list_messages(
+                    session_id=self.session_id,
+                    agent_id=agent.agent_id,
+                )
+                if all_messages:
+                    self._latest_agent_message[agent.agent_id] = all_messages[-1]
 
             # Skip restoring messages when conversation is managed server-side
             if isinstance(agent, Agent) and agent.model.stateful:
@@ -288,10 +316,13 @@ def initialize(self, agent: "LocalAgent", **kwargs: Any) -> None:
                     self.session_id,
                 )
             else:
-                # Restore the agents messages array including the optional prepend messages
-                agent.messages = prepend_messages + [
-                    session_message.to_message() for session_message in session_messages
-                ]
+                # Restore the agents messages array including the optional prepend messages and the
+                # still-live pinned head that compaction pushed back to the front.
+                agent.messages = (
+                    pinned_messages
+                    + prepend_messages
+                    + [session_message.to_message() for session_message in session_messages]
+                )
 
                 # Fix broken session histories: https://github.com/strands-agents/harness-sdk/issues/859
                 agent.messages = self._fix_broken_tool_use(agent.messages)

strands-py/tests/strands/agent/test_conversation_manager.py

@@ -1120,6 +1120,31 @@ def test_pin_first_only_applies_once():
     assert "first" in [m["content"][0]["text"] for m in agent.messages]
 
 
+def test_pin_first_head_count_includes_protected_tool_pair_partner():
+    """``pinned_head_count`` must cover the tool-pair partner that ``is_pinned`` also protects (#4087).
+
+    Trimming keeps a pinned assistant(toolUse) message's matching toolResult even though the result was
+    never explicitly pinned. A resumed session reattaches exactly ``pinned_head_count`` stored messages,
+    so that count has to include the partner or the restore drops the real tool result.
+    """
+    manager = SlidingWindowConversationManager(window_size=2, should_truncate_results=False, pin_first=2)
+    messages = [
+        {"role": "user", "content": [{"text": "use the tool"}]},
+        {"role": "assistant", "content": [{"toolUse": {"toolUseId": "tu1", "name": "echo", "input": {}}}]},
+        {
+            "role": "user",
+            "content": [
+                {"toolResult": {"toolUseId": "tu1", "status": "success", "content": [{"text": "REAL RESULT"}]}}
+            ],
+        },
+    ]
+    agent = _make_mock_agent(messages=messages)
+    manager.reduce_context(agent)
+
+    assert manager.pinned_head_count == 3
+    assert agent.messages[2]["content"][0]["toolResult"]["content"][0]["text"] == "REAL RESULT"
+
+
 def test_pinned_message_in_middle_survives_trimming():
     from strands.agent.conversation_manager.compression.pin_message import pin_message