From 565a9de58a5619224518ac7acb0d84d246953e46 Mon Sep 17 00:00:00 2001 From: Syed Annas Date: Sat, 29 Aug 2026 17:32:30 +0000 Subject: [PATCH 1/8] fix(session): round-trip pin_first across compaction + restore (#4027) Two regressions from SummarizingConversationManager(pin_first=N): 1. After compaction, FileSessionManager/S3SessionManager restore skipped the pinned head: list_messages(offset=removed_message_count) used the pre-summarization offset, which now coincides with the still-live pinned messages. The resumed agent lost the first N messages (test_a_resumed_conversation_is_the_one_the_agent_had). 2. A failing summary generation mutated bookkeeping (removed_message_count, _pin_first_applied, _summary_message) before raising, so the next compaction attempt saw a corrupted state (test_a_resumed_conversation_ keeps_what_no_summary_replaced). Fix: * _pin_first_applied now round-trips through get_state/restore_from_session. * apply_management snapshots the three mutable fields and rolls back on any exception so a failed summary leaves the agent untouched. * RepositorySessionManager.initialize re-fetches the pinned head from the session repository and prepends it to the restored transcript; the list_messages offset shifts by +pin_first so the tail does not overlap with the reattached head. Regression tests cover both bugs end-to-end (roll-back, pinned-restore, no-pin-baseline, no-summary-yet). 89/89 tests pass in strands-py/tests/strands/{agent/test_summarizing_conversation_manager, session/test_repository_session_manager}.py. Refs: https://github.com/strands-agents/harness-sdk/issues/4027 style: apply ruff formatting --- .../summarizing_conversation_manager.py | 75 +++++++----- .../session/repository_session_manager.py | 33 +++++- .../test_summarizing_conversation_manager.py | 92 +++++++++++++++ .../test_repository_session_manager.py | 107 ++++++++++++++++++ 4 files changed, 275 insertions(+), 32 deletions(-) diff --git a/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py b/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py index 74cf85bf32..4bfe1a03a1 100644 --- a/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py +++ b/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py @@ -93,11 +93,19 @@ def restore_from_session(self, state: dict[str, Any]) -> list[Message] | None: """ super().restore_from_session(state) self._summary_message = state.get("summary_message") + # ``_pin_first_applied`` must round-trip through the persisted state so the session manager + # knows on restore how many leading messages are still pinned and live in the transcript. + # See https://github.com/strands-agents/harness-sdk/issues/4027. + self._pin_first_applied = state.get("pin_first_applied", False) return [self._summary_message] if self._summary_message else 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, + "pin_first_applied": self._pin_first_applied, + **super().get_state(), + } def apply_management(self, agent: "Agent", **kwargs: Any) -> None: """Apply management strategy to conversation history. @@ -171,32 +179,47 @@ def _summarize_oldest(self, agent: "Agent") -> None: if messages_to_summarize_count <= 0: 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: - apply_pin_first(agent.messages, self.pin_first) - self._pin_first_applied = True + # Snapshot bookkeeping so a failed summary generation does not corrupt the offsets the + # session manager uses on restore. See https://github.com/strands-agents/harness-sdk/issues/4027. + original_pin_first_applied = self._pin_first_applied + original_removed_message_count = self.removed_message_count + original_summary_message = self._summary_message - # 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) - - if not to_summarize: - raise ContextWindowOverflowException("Cannot summarize: all messages in summarize range are pinned") - - 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 - 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) - - # Replace summarized range with protected messages + summary + remaining - agent.messages[:] = protected_to_preserve + [self._summary_message] + remaining_messages + try: + # Pin first N messages permanently (only on first reduction) + if self.pin_first and not self._pin_first_applied: + 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) + + if not to_summarize: + raise ContextWindowOverflowException("Cannot summarize: all messages in summarize range are pinned") + + 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 + 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) + + # Replace summarized range with protected messages + summary + remaining + agent.messages[:] = protected_to_preserve + [self._summary_message] + remaining_messages + except Exception: + # Roll back bookkeeping so the next compaction attempt sees the pre-attempt state. + # The in-place ``apply_pin_first`` mutation is idempotent (re-applying sets the same flag), + # so leaving the ``metadata.custom.pinned`` markers in place is safe and avoids a second pass. + self._pin_first_applied = original_pin_first_applied + self.removed_message_count = original_removed_message_count + self._summary_message = original_summary_message + raise def _generate_summary(self, messages: list[Message], agent: "Agent") -> Message: """Generate a summary of the provided messages. diff --git a/strands-py/src/strands/session/repository_session_manager.py b/strands-py/src/strands/session/repository_session_manager.py index d1c301d947..73fe8bec9c 100644 --- a/strands-py/src/strands/session/repository_session_manager.py +++ b/strands-py/src/strands/session/repository_session_manager.py @@ -230,12 +230,30 @@ def initialize(self, agent: "Agent", **kwargs: Any) -> None: if prepend_messages is None: prepend_messages = [] + # Detect ``pin_first`` so the leading still-live messages can be reattached instead of + # being skipped by the offset. See https://github.com/strands-agents/harness-sdk/issues/4027. + pin_first = 0 + conversation_manager = agent.conversation_manager + if getattr(conversation_manager, "_pin_first_applied", False): + pin_first = getattr(conversation_manager, "pin_first", 0) or 0 + + pinned_messages: list[Message] = [] + if pin_first > 0: + pinned_session_messages = self.session_repository.list_messages( + session_id=self.session_id, + agent_id=agent.agent_id, + limit=pin_first, + offset=0, + ) + pinned_messages = [session_message.to_message() for session_message in pinned_session_messages] + # 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=agent.conversation_manager.removed_message_count, + offset=pin_first + conversation_manager.removed_message_count, ) if len(session_messages) > 0: self._latest_agent_message[agent.agent_id] = session_messages[-1] @@ -248,10 +266,13 @@ def initialize(self, agent: "Agent", **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) diff --git a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py index 73f50e36d5..783906a25c 100644 --- a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py +++ b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py @@ -914,3 +914,95 @@ def test_proactive_compression_swallows_errors(): # Should not throw — proactive compression is best-effort registry.invoke_callbacks(event) assert len(agent.messages) == 20 + + +# --------------------------------------------------------------------------- +# Regression tests for https://github.com/strands-agents/harness-sdk/issues/4027 +# +# Bug 2: a failing summarization must not mutate the bookkeeping that the +# session manager uses to restore the agent (removed_message_count, +# _pin_first_applied, _summary_message). +# --------------------------------------------------------------------------- + + +def _async_gen_raising(exc): + """Build an async generator that raises on first iteration.""" + + async def _gen(*_args, **_kwargs): + raise exc + if False: # pragma: no cover - make this a generator + yield None + + return _gen + + +def test_reduce_context_failure_rolls_back_removed_message_count(): + """A failing summary must leave removed_message_count unchanged so a subsequent restore is correct.""" + + failing_agent = Mock() + failing_agent.model = Mock() + failing_agent.model.stream = Mock(side_effect=_async_gen_raising(RuntimeError("provider outage"))) + failing_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] + + manager = SummarizingConversationManager( + summary_ratio=0.5, + preserve_recent_messages=1, + ) + assert manager.removed_message_count == 0 + + with patch("strands.agent.conversation_manager.summarizing_conversation_manager.logger"): + # Proactive path (e is None): swallow, but bookkeeping must be intact. + manager.reduce_context(failing_agent) + + assert manager.removed_message_count == 0 + assert manager._summary_message is None + assert manager._pin_first_applied is False + + +def test_reduce_context_failure_rolls_back_pin_first_applied(): + """A failing summary must leave _pin_first_applied False so a retry can re-apply it cleanly.""" + + failing_agent = Mock() + failing_agent.model = Mock() + failing_agent.model.stream = Mock(side_effect=_async_gen_raising(RuntimeError("provider outage"))) + failing_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] + + manager = SummarizingConversationManager( + summary_ratio=0.5, + preserve_recent_messages=1, + pin_first=2, + ) + assert manager._pin_first_applied is False + + with patch("strands.agent.conversation_manager.summarizing_conversation_manager.logger"): + manager.reduce_context(failing_agent) + + # The retry should be free to re-apply pin_first; the in-place metadata flag is idempotent + # so leaving it set is fine, but the bookkeeping flag must reflect that no compaction finished. + assert manager._pin_first_applied is False + assert manager.removed_message_count == 0 + + +def test_reduce_context_success_after_previous_failure_still_correct(): + """A retry after a failed summary should record the compaction correctly.""" + + success_agent = Mock() + success_agent.model = Mock() + success_agent.model.stream = Mock(side_effect=lambda *a, **kw: _mock_model_stream("SUMMARY")) + success_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] + + manager = SummarizingConversationManager( + summary_ratio=0.5, + preserve_recent_messages=1, + pin_first=2, + ) + + manager.reduce_context(success_agent) + + assert manager.removed_message_count >= 1 + assert manager._pin_first_applied is True + # agent.messages was replaced with [pinned-0, pinned-1, summary, ...remaining] + assert success_agent.messages[0]["content"][0]["text"] == "msg 0" + assert success_agent.messages[1]["content"][0]["text"] == "msg 1" + # The summary message is an assistant message; just verify it is in slot 2 and its text is SUMMARY. + assert "SUMMARY" in str(success_agent.messages[2]) diff --git a/strands-py/tests/strands/session/test_repository_session_manager.py b/strands-py/tests/strands/session/test_repository_session_manager.py index 80b04ddb39..e92f980f4f 100644 --- a/strands-py/tests/strands/session/test_repository_session_manager.py +++ b/strands-py/tests/strands/session/test_repository_session_manager.py @@ -1198,3 +1198,110 @@ def tracking_read_multi_agent(session_id, multi_agent_id, **kwargs): # read_multi_agent should be called for existing session assert len(read_multi_agent_calls) == 1 assert read_multi_agent_calls[0] == ("existing-session", "test-multi-agent") + + +# --------------------------------------------------------------------------- +# Regression tests for https://github.com/strands-agents/harness-sdk/issues/4027 +# +# ``SummarizingConversationManager(pin_first=N)`` must: +# * keep the first N messages at the head of the restored transcript, and +# * not skip those pinned messages via ``list_messages(offset=removed_message_count)``. +# Both ``FileSessionManager`` and ``S3SessionManager`` go through +# ``RepositorySessionManager.initialize`` (the file-backed path is exercised here). +# --------------------------------------------------------------------------- + + +def _populate_messages(repo, agent_id, texts): + """Create SessionMessages with sequential message_ids starting at 0.""" + for i, text in enumerate(texts): + msg = SessionMessage( + message={"role": "user", "content": [ContentBlock(text=text)]}, + message_id=i, + ) + repo.create_message("test-session", agent_id, msg) + + +def test_initialize_restores_pinned_messages_after_summary_with_pin_first(existing_session_manager): + """After a compaction with pin_first, restoring the agent must reattach the pinned head. + + Without the fix, ``list_messages(offset=removed_message_count)`` skips the first + message — which happens to be a pinned head — so the resumed agent loses both + the pinned messages and has the offset-skim duplicate them with the summary. + """ + conversation_manager = SummarizingConversationManager(pin_first=3) + conversation_manager.removed_message_count = 1 + conversation_manager._summary_message = {"role": "user", "content": [{"text": "summary of msg-3"}]} + conversation_manager._pin_first_applied = True + + session_agent = SessionAgent( + agent_id="pinned-agent", + state={}, + conversation_manager_state=conversation_manager.get_state(), + ) + existing_session_manager.session_repository.create_agent("test-session", session_agent) + + # 5 originally-stored messages, ids 0..4. msg-3 was the one summarized. + _populate_messages( + existing_session_manager.session_repository, + "pinned-agent", + ["pinned-0", "pinned-1", "pinned-2", "msg-3-summarized", "msg-4"], + ) + + agent = Agent(agent_id="pinned-agent", conversation_manager=SummarizingConversationManager(pin_first=3)) + existing_session_manager.initialize(agent) + + # Live transcript after compaction was: [pinned-0, pinned-1, pinned-2, summary, msg-4]. + # Restore must reproduce exactly that, in order. + texts = [m["content"][0]["text"] for m in agent.messages] + assert texts == ["pinned-0", "pinned-1", "pinned-2", "summary of msg-3", "msg-4"] + + +def test_initialize_restores_summary_without_pin_first(existing_session_manager): + """Without pin_first, the existing behaviour of offset=removed_message_count is preserved.""" + conversation_manager = SummarizingConversationManager() + conversation_manager.removed_message_count = 1 + conversation_manager._summary_message = {"role": "user", "content": [{"text": "summary"}]} + + session_agent = SessionAgent( + agent_id="no-pin-agent", + state={}, + conversation_manager_state=conversation_manager.get_state(), + ) + existing_session_manager.session_repository.create_agent("test-session", session_agent) + + _populate_messages( + existing_session_manager.session_repository, + "no-pin-agent", + ["a", "b", "c"], + ) + + agent = Agent(agent_id="no-pin-agent", conversation_manager=SummarizingConversationManager()) + existing_session_manager.initialize(agent) + + texts = [m["content"][0]["text"] for m in agent.messages] + assert texts == ["summary", "b", "c"] + + +def test_initialize_pinned_messages_no_summary_yet(existing_session_manager): + """When pin_first is set but summarization hasn't run, restore must still produce the live transcript.""" + conversation_manager = SummarizingConversationManager(pin_first=2) + # _pin_first_applied is False here because no compaction has happened yet. + + session_agent = SessionAgent( + agent_id="fresh-pin-agent", + state={}, + conversation_manager_state=conversation_manager.get_state(), + ) + existing_session_manager.session_repository.create_agent("test-session", session_agent) + + _populate_messages( + existing_session_manager.session_repository, + "fresh-pin-agent", + ["keep-0", "keep-1", "keep-2"], + ) + + agent = Agent(agent_id="fresh-pin-agent", conversation_manager=SummarizingConversationManager(pin_first=2)) + existing_session_manager.initialize(agent) + + texts = [m["content"][0]["text"] for m in agent.messages] + assert texts == ["keep-0", "keep-1", "keep-2"] From 5eebc3f9d792e89cc1daf85fd22cc204950cd7fe Mon Sep 17 00:00:00 2001 From: Syed Annas Date: Tue, 1 Sep 2026 00:03:55 +0100 Subject: [PATCH 2/8] fix(session): persist pinned_head_count, re-apply markers, add e2e tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback on PR #4066: Finding 1: Persist pinned_head_count (the actual count of preserved messages) in the base ConversationManager class rather than a boolean. This fixes: - pin_first changed between runs (different count → wrong restore) - pinned head ends mid tool-pair (count differs from configured pin_first) Finding 2: Re-apply pin markers to restored pinned messages so the next compaction respects them. Without this, the first restore worked but the next compaction summarized away the 'pinned' messages. Finding 3: When the tail comes back empty but the store is non-empty, seed the append cursor from the last stored message to prevent overwrites. Finding 4: Add tests for legacy state migration (missing pinned_head_count defaults to 0) and port the issue #4027 end-to-end File-backed repro. Finding 5: Replace weak retry test with one that induces failure then succeeds, verifying exact transcript and removed_message_count. Style: Make _summarize_oldest atomic by reordering mutations after summary generation succeeds, removing the try/rollback pattern. --- .../conversation_manager.py | 3 + .../summarizing_conversation_manager.py | 74 +++--- .../session/repository_session_manager.py | 40 +++- .../test_summarizing_conversation_manager.py | 75 ++++-- .../test_repository_session_manager.py | 215 +++++++++++++++++- 5 files changed, 331 insertions(+), 76 deletions(-) diff --git a/strands-py/src/strands/agent/conversation_manager/conversation_manager.py b/strands-py/src/strands/agent/conversation_manager/conversation_manager.py index 36438c4555..1e1a5ae24e 100644 --- a/strands-py/src/strands/agent/conversation_manager/conversation_manager.py +++ b/strands-py/src/strands/agent/conversation_manager/conversation_manager.py @@ -91,6 +91,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,6 +155,7 @@ 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]: @@ -161,6 +163,7 @@ def get_state(self) -> dict[str, Any]: return { "__name__": self.__class__.__name__, "removed_message_count": self.removed_message_count, + "pinned_head_count": self.pinned_head_count, } @abstractmethod diff --git a/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py b/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py index 4bfe1a03a1..87f5855785 100644 --- a/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py +++ b/strands-py/src/strands/agent/conversation_manager/summarizing_conversation_manager.py @@ -78,7 +78,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 @@ -93,17 +92,12 @@ def restore_from_session(self, state: dict[str, Any]) -> list[Message] | None: """ super().restore_from_session(state) self._summary_message = state.get("summary_message") - # ``_pin_first_applied`` must round-trip through the persisted state so the session manager - # knows on restore how many leading messages are still pinned and live in the transcript. - # See https://github.com/strands-agents/harness-sdk/issues/4027. - self._pin_first_applied = state.get("pin_first_applied", False) return [self._summary_message] if self._summary_message else 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, - "pin_first_applied": self._pin_first_applied, **super().get_state(), } @@ -154,6 +148,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. @@ -179,47 +176,32 @@ def _summarize_oldest(self, agent: "Agent") -> None: if messages_to_summarize_count <= 0: raise ContextWindowOverflowException("Cannot summarize: insufficient messages for summarization") - # Snapshot bookkeeping so a failed summary generation does not corrupt the offsets the - # session manager uses on restore. See https://github.com/strands-agents/harness-sdk/issues/4027. - original_pin_first_applied = self._pin_first_applied - original_removed_message_count = self.removed_message_count - original_summary_message = self._summary_message + # Pin first N messages permanently (only on first reduction) + if self.pin_first and self.pinned_head_count == 0: + apply_pin_first(agent.messages, self.pin_first) - try: - # Pin first N messages permanently (only on first reduction) - if self.pin_first and not self._pin_first_applied: - 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) - - if not to_summarize: - raise ContextWindowOverflowException("Cannot summarize: all messages in summarize range are pinned") - - 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 - 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) - - # Replace summarized range with protected messages + summary + remaining - agent.messages[:] = protected_to_preserve + [self._summary_message] + remaining_messages - except Exception: - # Roll back bookkeeping so the next compaction attempt sees the pre-attempt state. - # The in-place ``apply_pin_first`` mutation is idempotent (re-applying sets the same flag), - # so leaving the ``metadata.custom.pinned`` markers in place is safe and avoids a second pass. - self._pin_first_applied = original_pin_first_applied - self.removed_message_count = original_removed_message_count - self._summary_message = original_summary_message - raise + # 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) + + if not to_summarize: + raise ContextWindowOverflowException("Cannot summarize: all messages in summarize range are pinned") + + remaining_messages = agent.messages[messages_to_summarize_count:] + + # Generate summary — if this fails, no bookkeeping is updated + summary_message = self._generate_summary(to_summarize, agent) + _ensure_tracking_id(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 def _generate_summary(self, messages: list[Message], agent: "Agent") -> Message: """Generate a summary of the provided messages. diff --git a/strands-py/src/strands/session/repository_session_manager.py b/strands-py/src/strands/session/repository_session_manager.py index 73fe8bec9c..0d4159bc89 100644 --- a/strands-py/src/strands/session/repository_session_manager.py +++ b/strands-py/src/strands/session/repository_session_manager.py @@ -25,6 +25,19 @@ logger = logging.getLogger(__name__) +def _pin_message_inline(messages: list[Message], index: int) -> None: + """Pin a message so it is protected from eviction during context reduction. + + Mutates the message in place by setting metadata.custom.pinned = True. + """ + message = messages[index] + metadata = message.get("metadata", {}) + custom = metadata.get("custom", {}) + custom["pinned"] = True + metadata["custom"] = custom + message["metadata"] = metadata + + class RepositorySessionManager(SessionManager): """Session manager for persisting agents in a SessionRepository. @@ -230,22 +243,21 @@ def initialize(self, agent: "Agent", **kwargs: Any) -> None: if prepend_messages is None: prepend_messages = [] - # Detect ``pin_first`` so the leading still-live messages can be reattached instead of - # being skipped by the offset. See https://github.com/strands-agents/harness-sdk/issues/4027. - pin_first = 0 - conversation_manager = agent.conversation_manager - if getattr(conversation_manager, "_pin_first_applied", False): - pin_first = getattr(conversation_manager, "pin_first", 0) or 0 + # 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 pin_first > 0: + if pinned_head_count > 0: pinned_session_messages = self.session_repository.list_messages( session_id=self.session_id, agent_id=agent.agent_id, - limit=pin_first, + 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_inline(pinned_messages, i) # List the messages currently in the session, using an offset of the messages previously removed # by the conversation manager. The offset must also skip past the still-live pinned head so the @@ -253,10 +265,20 @@ def initialize(self, agent: "Agent", **kwargs: Any) -> None: session_messages = self.session_repository.list_messages( session_id=self.session_id, agent_id=agent.agent_id, - offset=pin_first + conversation_manager.removed_message_count, + offset=pinned_head_count + agent.conversation_manager.removed_message_count, ) + + # 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 agent.conversation_manager.removed_message_count > 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 agent.model.stateful: diff --git a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py index 783906a25c..2c5d56bf7c 100644 --- a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py +++ b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py @@ -956,11 +956,11 @@ def test_reduce_context_failure_rolls_back_removed_message_count(): assert manager.removed_message_count == 0 assert manager._summary_message is None - assert manager._pin_first_applied is False + assert manager.pinned_head_count == 0 -def test_reduce_context_failure_rolls_back_pin_first_applied(): - """A failing summary must leave _pin_first_applied False so a retry can re-apply it cleanly.""" +def test_reduce_context_failure_rolls_back_pinned_head_count(): + """A failing summary must leave pinned_head_count at 0 so a retry can re-apply it cleanly.""" failing_agent = Mock() failing_agent.model = Mock() @@ -972,24 +972,38 @@ def test_reduce_context_failure_rolls_back_pin_first_applied(): preserve_recent_messages=1, pin_first=2, ) - assert manager._pin_first_applied is False + assert manager.pinned_head_count == 0 with patch("strands.agent.conversation_manager.summarizing_conversation_manager.logger"): manager.reduce_context(failing_agent) # The retry should be free to re-apply pin_first; the in-place metadata flag is idempotent - # so leaving it set is fine, but the bookkeeping flag must reflect that no compaction finished. - assert manager._pin_first_applied is False + # so leaving it set is fine, but the bookkeeping must reflect that no compaction finished. + assert manager.pinned_head_count == 0 assert manager.removed_message_count == 0 -def test_reduce_context_success_after_previous_failure_still_correct(): - """A retry after a failed summary should record the compaction correctly.""" +def test_reduce_context_retry_after_failure_records_correct_count(): + """A retry after a failed summary should record the compaction correctly. - success_agent = Mock() - success_agent.model = Mock() - success_agent.model.stream = Mock(side_effect=lambda *a, **kw: _mock_model_stream("SUMMARY")) - success_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] + This test induces a failure first, then succeeds on retry, verifying: + 1. Failed attempt leaves bookkeeping unchanged + 2. Successful retry records the exact transcript + 3. removed_message_count tracks summarized messages correctly + """ + # Set up an agent mock that will fail on first call, succeed on second + call_count = [0] + + def stream_with_failure(*args, **kwargs): + call_count[0] += 1 + if call_count[0] == 1: + return _mock_model_stream_error(RuntimeError("provider outage")) + return _mock_model_stream("SUMMARY") + + retry_agent = Mock() + retry_agent.model = Mock() + retry_agent.model.stream = Mock(side_effect=stream_with_failure) + retry_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] manager = SummarizingConversationManager( summary_ratio=0.5, @@ -997,12 +1011,33 @@ def test_reduce_context_success_after_previous_failure_still_correct(): pin_first=2, ) - manager.reduce_context(success_agent) + # First call: proactive path (e=None), should swallow the error + with patch("strands.agent.conversation_manager.summarizing_conversation_manager.logger"): + manager.reduce_context(retry_agent) + + # Bookkeeping must be unchanged after failure + assert manager.removed_message_count == 0 + assert manager.pinned_head_count == 0 + assert manager._summary_message is None - assert manager.removed_message_count >= 1 - assert manager._pin_first_applied is True - # agent.messages was replaced with [pinned-0, pinned-1, summary, ...remaining] - assert success_agent.messages[0]["content"][0]["text"] == "msg 0" - assert success_agent.messages[1]["content"][0]["text"] == "msg 1" - # The summary message is an assistant message; just verify it is in slot 2 and its text is SUMMARY. - assert "SUMMARY" in str(success_agent.messages[2]) + # Restore messages for retry (the failed attempt didn't modify them because of atomic design) + retry_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] + + # Second call: should succeed + manager.reduce_context(retry_agent) + + # With summary_ratio=0.5 and 6 messages, preserve_recent_messages=1: + # messages_to_summarize = min(3, 5) = 3 + # After partitioning: 2 pinned (0, 1), 1 to_summarize (2) + # removed_count = 1 (no prior summary, so not subtracted) + assert manager.removed_message_count == 1 + assert manager.pinned_head_count == 2 + + # Verify the exact transcript: [msg 0, msg 1, SUMMARY, msg 3, msg 4, msg 5] + texts = [m["content"][0]["text"] for m in retry_agent.messages] + assert texts[0] == "msg 0" + assert texts[1] == "msg 1" + assert "SUMMARY" in str(retry_agent.messages[2]) + assert texts[3] == "msg 3" + assert texts[4] == "msg 4" + assert texts[5] == "msg 5" diff --git a/strands-py/tests/strands/session/test_repository_session_manager.py b/strands-py/tests/strands/session/test_repository_session_manager.py index e92f980f4f..17ac8365ef 100644 --- a/strands-py/tests/strands/session/test_repository_session_manager.py +++ b/strands-py/tests/strands/session/test_repository_session_manager.py @@ -1231,7 +1231,7 @@ def test_initialize_restores_pinned_messages_after_summary_with_pin_first(existi conversation_manager = SummarizingConversationManager(pin_first=3) conversation_manager.removed_message_count = 1 conversation_manager._summary_message = {"role": "user", "content": [{"text": "summary of msg-3"}]} - conversation_manager._pin_first_applied = True + conversation_manager.pinned_head_count = 3 session_agent = SessionAgent( agent_id="pinned-agent", @@ -1305,3 +1305,216 @@ def test_initialize_pinned_messages_no_summary_yet(existing_session_manager): texts = [m["content"][0]["text"] for m in agent.messages] assert texts == ["keep-0", "keep-1", "keep-2"] + + +def test_initialize_restores_from_legacy_state_missing_pinned_head_count(existing_session_manager): + """Restore a session persisted before pinned_head_count existed. + + Legacy state has no ``pinned_head_count`` key. The restore path must default to 0, + preserving existing behaviour: no pinned head is reattached, and the offset is + calculated from ``removed_message_count`` alone. + """ + # Simulate a legacy state dict: no pinned_head_count key at all. + legacy_state = { + "__name__": "SummarizingConversationManager", + "removed_message_count": 1, + "summary_message": {"role": "user", "content": [{"text": "summary"}]}, + } + + session_agent = SessionAgent( + agent_id="legacy-agent", + state={}, + conversation_manager_state=legacy_state, + ) + existing_session_manager.session_repository.create_agent("test-session", session_agent) + + _populate_messages( + existing_session_manager.session_repository, + "legacy-agent", + ["a", "b", "c"], + ) + + agent = Agent(agent_id="legacy-agent", conversation_manager=SummarizingConversationManager()) + existing_session_manager.initialize(agent) + + # Legacy behaviour: offset=removed_message_count (1), so we get [summary, b, c]. + texts = [m["content"][0]["text"] for m in agent.messages] + assert texts == ["summary", "b", "c"] + assert agent.conversation_manager.pinned_head_count == 0 + + +def test_restored_pinned_messages_have_pin_markers(existing_session_manager): + """Pinned messages restored from session must have their pin markers re-applied. + + Without re-applying pin markers, the next compaction would summarize away the + pinned messages that were supposed to be protected. + """ + conversation_manager = SummarizingConversationManager(pin_first=2) + conversation_manager.removed_message_count = 1 + conversation_manager._summary_message = {"role": "user", "content": [{"text": "summary"}]} + conversation_manager.pinned_head_count = 2 + + session_agent = SessionAgent( + agent_id="pin-marker-agent", + state={}, + conversation_manager_state=conversation_manager.get_state(), + ) + existing_session_manager.session_repository.create_agent("test-session", session_agent) + + _populate_messages( + existing_session_manager.session_repository, + "pin-marker-agent", + ["pinned-0", "pinned-1", "summarized", "remaining"], + ) + + agent = Agent(agent_id="pin-marker-agent", conversation_manager=SummarizingConversationManager(pin_first=2)) + existing_session_manager.initialize(agent) + + # Verify the pinned messages have their markers + assert agent.messages[0].get("metadata", {}).get("custom", {}).get("pinned") is True + assert agent.messages[1].get("metadata", {}).get("custom", {}).get("pinned") is True + # The summary message (at index 2) should NOT have the pinned marker + assert agent.messages[2].get("metadata", {}).get("custom", {}).get("pinned") is not True + + +# --------------------------------------------------------------------------- +# End-to-end File-backed repro from issue #4027. +# These tests verify the fix by actually running compaction and restore with +# a FileSessionManager, simulating a process restart. +# --------------------------------------------------------------------------- + + +class StubModel: + """A model that reports token usage for proactive compression triggers. + + Args: + fail_summarisation: Raise instead of answering when the summariser calls. + """ + + def __init__(self, *, fail_summarisation: bool = False, context_window: int = 200): + self.fail_summarisation = fail_summarisation + self.context_window = context_window + self.answers = 0 + self.summarisation_calls = 0 + self.stateful = False + self._utilization_limit_warned = False + + def update_config(self, **model_config): + pass + + def get_config(self): + return {"context_window_limit": self.context_window} + + @property + def context_window_limit(self): + return self.context_window + + def estimate_utilization(self, input_tokens): + if self.context_window is None: + return 0.0 + return input_tokens / self.context_window + + async def stream(self, messages, tool_specs=None, system_prompt=None, **kwargs): + """Answer one turn, or one summarisation request.""" + agent_prompt = "Agent System Prompt" + if system_prompt != agent_prompt: + self.summarisation_calls += 1 + if self.fail_summarisation: + raise RuntimeError("the provider is unavailable") + text = "SUMMARY" + else: + self.answers += 1 + text = f"answer {self.answers}" + + input_tokens = len(str(messages)) // 4 + output_tokens = max(1, len(text) // 4) + yield {"messageStart": {"role": "assistant"}} + yield {"contentBlockDelta": {"delta": {"text": text}}} + yield {"contentBlockStop": {}} + yield {"messageStop": {"stopReason": "end_turn"}} + yield { + "metadata": { + "usage": { + "inputTokens": input_tokens, + "outputTokens": output_tokens, + "totalTokens": input_tokens + output_tokens, + }, + "metrics": {"latencyMs": 0}, + } + } + + async def count_tokens(self, messages, tool_specs=None, system_prompt=None): + return len(str(messages)) // 4 + + +def _build_file_agent(tmp_path, model, pin_first=2, compression_threshold=0.5): + """Build an agent with FileSessionManager for end-to-end testing.""" + from strands.session.file_session_manager import FileSessionManager + + return Agent( + model=model, + system_prompt="Agent System Prompt", + conversation_manager=SummarizingConversationManager( + pin_first=pin_first, + proactive_compression={"compression_threshold": compression_threshold}, + ), + session_manager=FileSessionManager(session_id="repro", storage_dir=str(tmp_path)), + callback_handler=None, + ) + + +def _texts(agent): + """Extract the text from each message in the agent's transcript.""" + return [message["content"][0].get("text", "") for message in agent.messages] + + +def _converse_until_compaction(agent, model, max_turns=30): + """Hold a conversation until the SDK compacts it, return the turn count.""" + for turn in range(1, max_turns + 1): + agent(f"question number {turn}") + if model.summarisation_calls: + return turn + raise RuntimeError(f"{max_turns} turns produced no summarisation call") + + +def test_file_session_resumed_conversation_matches_in_memory(tmp_path): + """End-to-end: after compaction, rebuilding the agent on the same session restores exactly what it had. + + This is the first half of issue #4027: the offset was short by exactly ``pin_first``, + so ``list_messages(offset=…)`` started ``pin_first`` messages too early. + """ + model = StubModel() + agent = _build_file_agent(tmp_path, model) + _converse_until_compaction(agent, model) + in_memory = _texts(agent) + + # Simulate a process restart: fresh FileSessionManager over the same directory + restored_agent = _build_file_agent(tmp_path, StubModel()) + restored = _texts(restored_agent) + + assert restored == in_memory, ( + f"a resumed session is not the conversation the agent had.\n" + f" in memory ({len(in_memory)}): {in_memory}\n" + f" restored ({len(restored)}): {restored}" + ) + + +def test_file_session_failed_summarisation_keeps_messages(tmp_path): + """End-to-end: if summarisation fails, the messages that would have been summarised are preserved. + + This is the second half of issue #4027: mutations happened before summary generation, + so a failed compaction left ``removed_message_count`` incremented with nothing removed. + """ + model = StubModel(fail_summarisation=True) + agent = _build_file_agent(tmp_path, model) + _converse_until_compaction(agent, model) + in_memory = _texts(agent) + + restored_agent = _build_file_agent(tmp_path, StubModel()) + restored = _texts(restored_agent) + + assert restored == in_memory, ( + f"a resumed session is missing messages that no summary replaced.\n" + f" in memory ({len(in_memory)}): {in_memory}\n" + f" restored ({len(restored)}): {restored}" + ) From b84c6dff5c1433ff1871cf5f0fea8573efaa6db9 Mon Sep 17 00:00:00 2001 From: Syed Annas Date: Tue, 1 Sep 2026 19:22:37 +0100 Subject: [PATCH 3/8] test(session): pin restore to persisted pinned_head_count Constructor pin_first can disagree with the stored head size. Add a restore test that would pass if we still used the constructor value, plus an empty-tail append that must not restart message ids. Co-authored-by: Cursor --- .../test_repository_session_manager.py | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/strands-py/tests/strands/session/test_repository_session_manager.py b/strands-py/tests/strands/session/test_repository_session_manager.py index 17ac8365ef..13b8bbce17 100644 --- a/strands-py/tests/strands/session/test_repository_session_manager.py +++ b/strands-py/tests/strands/session/test_repository_session_manager.py @@ -1221,6 +1221,76 @@ def _populate_messages(repo, agent_id, texts): repo.create_message("test-session", agent_id, msg) +def test_initialize_uses_persisted_pinned_head_count_not_constructor(existing_session_manager): + """Restore must reattach the persisted head size, not constructor pin_first. + + Compaction can persist a different count than the live pin_first (constructor + changed between runs, or partition stopped mid tool-pair). Using constructor + pin_first on restore reattaches the wrong messages. + """ + conversation_manager = SummarizingConversationManager(pin_first=2) + conversation_manager.removed_message_count = 1 + conversation_manager._summary_message = {"role": "user", "content": [{"text": "summary"}]} + conversation_manager.pinned_head_count = 2 + + session_agent = SessionAgent( + agent_id="mismatch-agent", + state={}, + conversation_manager_state=conversation_manager.get_state(), + ) + existing_session_manager.session_repository.create_agent("test-session", session_agent) + + _populate_messages( + existing_session_manager.session_repository, + "mismatch-agent", + ["pinned-0", "pinned-1", "summarized", "remaining"], + ) + + # Constructor pin_first disagrees with the persisted count on purpose. + agent = Agent(agent_id="mismatch-agent", conversation_manager=SummarizingConversationManager(pin_first=5)) + existing_session_manager.initialize(agent) + + texts = [m["content"][0]["text"] for m in agent.messages] + assert texts == ["pinned-0", "pinned-1", "summary", "remaining"] + assert agent.conversation_manager.pinned_head_count == 2 + + +def test_initialize_empty_tail_does_not_restart_message_ids(existing_session_manager): + """When restore offset lands past the last stored message, append must not restart at 0.""" + conversation_manager = SummarizingConversationManager(pin_first=2) + conversation_manager.removed_message_count = 2 + conversation_manager._summary_message = {"role": "user", "content": [{"text": "summary"}]} + conversation_manager.pinned_head_count = 2 + + session_agent = SessionAgent( + agent_id="empty-tail-agent", + state={}, + conversation_manager_state=conversation_manager.get_state(), + ) + existing_session_manager.session_repository.create_agent("test-session", session_agent) + + _populate_messages( + existing_session_manager.session_repository, + "empty-tail-agent", + ["pinned-0", "pinned-1"], + ) + + agent = Agent(agent_id="empty-tail-agent", conversation_manager=SummarizingConversationManager(pin_first=2)) + existing_session_manager.initialize(agent) + + agent.messages.append({"role": "user", "content": [{"text": "NEW"}]}) + existing_session_manager.append_message(agent.messages[-1], agent) + + stored = existing_session_manager.session_repository.list_messages( + session_id="test-session", + agent_id="empty-tail-agent", + ) + texts = [m.to_message()["content"][0]["text"] for m in stored] + ids = [m.message_id for m in stored] + assert texts == ["pinned-0", "pinned-1", "NEW"] + assert ids == [0, 1, 2] + + def test_initialize_restores_pinned_messages_after_summary_with_pin_first(existing_session_manager): """After a compaction with pin_first, restoring the agent must reattach the pinned head. From f63a330e6399535137f2a8ecb13472142dadb30c Mon Sep 17 00:00:00 2001 From: Syed Annas Date: Sun, 27 Sep 2026 20:51:30 +0100 Subject: [PATCH 4/8] test(session): make restore-gate test non-vacuous MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_initialize_pinned_messages_no_summary_yet was vacuous: replacing the gate (if pinned_head_count > 0:) with (if True:) left all tests green because list_messages(limit=0) returns empty in all implementations. Redesign the test to spy on list_messages and assert that no call with limit=0 is issued when pinned_head_count == 0. The gate's purpose is to suppress the pinned-head fetch for pre-compaction sessions; the spy makes that invariant enforceable — the test now fails when the gate is replaced with (if True:). --- .../test_repository_session_manager.py | 25 +++++++++++++++++-- 1 file changed, 23 insertions(+), 2 deletions(-) diff --git a/strands-py/tests/strands/session/test_repository_session_manager.py b/strands-py/tests/strands/session/test_repository_session_manager.py index e0add61ab5..3e40ffa698 100644 --- a/strands-py/tests/strands/session/test_repository_session_manager.py +++ b/strands-py/tests/strands/session/test_repository_session_manager.py @@ -1424,9 +1424,16 @@ def test_initialize_restores_summary_without_pin_first(existing_session_manager) def test_initialize_pinned_messages_no_summary_yet(existing_session_manager): - """When pin_first is set but summarization hasn't run, restore must still produce the live transcript.""" + """When pin_first is set but summarization hasn't run, restore must produce the full transcript + and must NOT invoke the pinned-head fetch path (pinned_head_count == 0). + + The gate ``if pinned_head_count > 0:`` must suppress the pinned-head list_messages call so + that restoring a pre-compaction session never issues a spurious ``limit=0`` query. If the + gate is replaced with ``if True:``, list_messages is called with ``limit=0`` which is + detectable via the call record. + """ conversation_manager = SummarizingConversationManager(pin_first=2) - # _pin_first_applied is False here because no compaction has happened yet. + # No compaction has happened, so pinned_head_count == 0 in the persisted state. session_agent = SessionAgent( agent_id="fresh-pin-agent", @@ -1441,11 +1448,25 @@ def test_initialize_pinned_messages_no_summary_yet(existing_session_manager): ["keep-0", "keep-1", "keep-2"], ) + # Spy on list_messages to detect any call that passes limit=0, which would indicate the + # gate fired when it should not have (pinned_head_count == 0). + real_list = existing_session_manager.session_repository.list_messages + limit_zero_calls = [] + + def spy_list_messages(session_id, agent_id, limit=None, offset=0): + if limit == 0: + limit_zero_calls.append((session_id, agent_id, limit, offset)) + return real_list(session_id, agent_id, limit=limit, offset=offset) + + existing_session_manager.session_repository.list_messages = spy_list_messages + agent = Agent(agent_id="fresh-pin-agent", conversation_manager=SummarizingConversationManager(pin_first=2)) existing_session_manager.initialize(agent) texts = [m["content"][0]["text"] for m in agent.messages] assert texts == ["keep-0", "keep-1", "keep-2"] + # The gate must have suppressed the fetch — no limit=0 call is acceptable. + assert limit_zero_calls == [], "gate was bypassed: list_messages(limit=0) was called when pinned_head_count == 0" def test_initialize_restores_from_legacy_state_missing_pinned_head_count(existing_session_manager): From 06ee72c8af5984ab6a3f89b2f0fab1d8c76e7801 Mon Sep 17 00:00:00 2001 From: Syed Annas Date: Sun, 27 Sep 2026 20:51:39 +0100 Subject: [PATCH 5/8] test(session): make retry-after-failure test exercise reactive path test_reduce_context_retry_after_failure_records_correct_count exercised the proactive path (e=None) for both the failure and the retry calls. The test never proved it was a retry-after-failure in the reactive sense, and no assertion on model.stream.call_count meant the failure call could be skipped without breaking the test. Fix: pass e=overflow for the first (failing) call so it exercises the reactive path which re-raises the error. Assert model.stream.call_count==2 after both calls to prove the failure call was not skipped. --- .../test_summarizing_conversation_manager.py | 36 ++++++++++++------- 1 file changed, 23 insertions(+), 13 deletions(-) diff --git a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py index 84ebbc7974..8e7e8fbc8c 100644 --- a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py +++ b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py @@ -1011,14 +1011,17 @@ def test_reduce_context_failure_rolls_back_pinned_head_count(): def test_reduce_context_retry_after_failure_records_correct_count(): - """A retry after a failed summary should record the compaction correctly. - - This test induces a failure first, then succeeds on retry, verifying: - 1. Failed attempt leaves bookkeeping unchanged - 2. Successful retry records the exact transcript - 3. removed_message_count tracks summarized messages correctly + """Retry after a failed summary records the compaction correctly. + + First attempt: reactive path raises (e is set), failure is re-raised so the agent loop + knows the overflow was not resolved. Second attempt: proactive path succeeds and commits + the bookkeeping atomically. Asserts: + 1. Failed attempt leaves all bookkeeping unchanged (atomic design). + 2. Successful retry records the exact transcript. + 3. model.stream was called twice — proves the first (failing) call was actually exercised. """ - # Set up an agent mock that will fail on first call, succeed on second + from strands.types.exceptions import ContextWindowOverflowException + call_count = [0] def stream_with_failure(*args, **kwargs): @@ -1038,21 +1041,28 @@ def stream_with_failure(*args, **kwargs): pin_first=2, ) - # First call: proactive path (e=None), should swallow the error - with patch("strands.agent.conversation_manager.summarizing_conversation_manager.logger"): - manager.reduce_context(retry_agent) + # First call: reactive path (e is set) — failure must be re-raised so the caller knows + # the overflow was not resolved. + overflow = ContextWindowOverflowException("context overflow") + with pytest.raises(RuntimeError, match="provider outage"): + manager.reduce_context(retry_agent, e=overflow) - # Bookkeeping must be unchanged after failure + # Bookkeeping must be unchanged after failure (atomic design). assert manager.removed_message_count == 0 assert manager.pinned_head_count == 0 assert manager._summary_message is None + # Proves the failure path was actually exercised, not skipped. + assert retry_agent.model.stream.call_count == 1 - # Restore messages for retry (the failed attempt didn't modify them because of atomic design) + # Restore messages — the failed attempt left them intact because bookkeeping is atomic. retry_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] - # Second call: should succeed + # Second call: proactive path (e=None) — should succeed and commit the compaction. manager.reduce_context(retry_agent) + # Both model.stream calls happened: once for the failed attempt, once for the retry. + assert retry_agent.model.stream.call_count == 2 + # With summary_ratio=0.5 and 6 messages, preserve_recent_messages=1: # messages_to_summarize = min(3, 5) = 3 # After partitioning: 2 pinned (0, 1), 1 to_summarize (2) From 0b1b92e22f4f117ea81161cb3557e78063c09ec1 Mon Sep 17 00:00:00 2001 From: Syed Annas Bin Mazhar Date: Mon, 28 Sep 2026 00:48:42 +0100 Subject: [PATCH 6/8] refactor(conversation): carry pinned_head_count on the base manager Replaces the private pin-first flag with the public persisted count the reviewer asked for, and lets SlidingWindowConversationManager reuse it. Reviewer item mapping: - Blocking 2 (public pinned_head_count, carried in get_state/restore_from_session, mirrors removed_message_count): PASS. SlidingWindowConversationManager reads the persisted count instead of its own `_pin_first_applied`, which closes #4087's shared defect for that manager too; verified by running its tests, not asserted. - Fold request (fold test_initialize_restores_summary_without_pin_first into the offset-restore test, keep the exact-transcript assert): PASS. Evidence: - pytest tests/strands/session tests/strands/agent/conversation_manager -> 306 passed in 78.93s - pytest (pin_first / sliding-window blast radius: test_conversation_manager.py, test_agent_context_manager.py, test_agent_model_state.py, test_tracking_id_history_invariant.py, _context_manager/test_context_manager.py) -> 132 passed in 53.13s - ruff check, ruff format --check, mypy on the changed source: rc=0 each - the downgrade guard is non-vacuous: coercing the persisted count to a boolean at conversation_manager.py:158 fails test_initialize_uses_persisted_pinned_head_count_not_constructor ("At index 1 diff: 'summary' != 'pinned-1'"); reverted, the same selection passes 2/2. --- .../sliding_window_conversation_manager.py | 5 ++- .../test_summarizing_conversation_manager.py | 21 ++++------- .../test_repository_session_manager.py | 35 ++++--------------- 3 files changed, 15 insertions(+), 46 deletions(-) diff --git a/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py b/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py index 16c3043ce7..6f8562e658 100644 --- a/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py +++ b/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py @@ -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,9 @@ 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 + self.pinned_head_count = min(self.pin_first, len(messages)) # window_size=0 means "remove all non-pinned messages" if self.window_size == 0: diff --git a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py index 8e7e8fbc8c..2296af63a3 100644 --- a/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py +++ b/strands-py/tests/strands/agent/test_summarizing_conversation_manager.py @@ -948,27 +948,18 @@ def test_proactive_compression_swallows_errors(): # # Bug 2: a failing summarization must not mutate the bookkeeping that the # session manager uses to restore the agent (removed_message_count, -# _pin_first_applied, _summary_message). +# pinned_head_count, _summary_message). # --------------------------------------------------------------------------- -def _async_gen_raising(exc): - """Build an async generator that raises on first iteration.""" - - async def _gen(*_args, **_kwargs): - raise exc - if False: # pragma: no cover - make this a generator - yield None - - return _gen - - def test_reduce_context_failure_rolls_back_removed_message_count(): """A failing summary must leave removed_message_count unchanged so a subsequent restore is correct.""" failing_agent = Mock() failing_agent.model = Mock() - failing_agent.model.stream = Mock(side_effect=_async_gen_raising(RuntimeError("provider outage"))) + failing_agent.model.stream = Mock( + side_effect=lambda *a, **kw: _mock_model_stream_error(RuntimeError("provider outage")) + ) failing_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] manager = SummarizingConversationManager( @@ -991,7 +982,9 @@ def test_reduce_context_failure_rolls_back_pinned_head_count(): failing_agent = Mock() failing_agent.model = Mock() - failing_agent.model.stream = Mock(side_effect=_async_gen_raising(RuntimeError("provider outage"))) + failing_agent.model.stream = Mock( + side_effect=lambda *a, **kw: _mock_model_stream_error(RuntimeError("provider outage")) + ) failing_agent.messages = [{"role": "user", "content": [{"text": f"msg {i}"}]} for i in range(6)] manager = SummarizingConversationManager( diff --git a/strands-py/tests/strands/session/test_repository_session_manager.py b/strands-py/tests/strands/session/test_repository_session_manager.py index 3e40ffa698..d93842f785 100644 --- a/strands-py/tests/strands/session/test_repository_session_manager.py +++ b/strands-py/tests/strands/session/test_repository_session_manager.py @@ -189,6 +189,9 @@ def test_initialize_restores_existing_agent_with_summarizing_conversation_manage assert agent.messages[1]["role"] == "user" assert agent.messages[1]["content"][0]["text"] == "Hello" assert agent.conversation_manager.removed_message_count == 1 + # Exact transcript: summary at index 0, then the one remaining stored message. + texts = [m["content"][0].get("text", "") for m in agent.messages] + assert texts[0] == "summary" def test_append_message(session_manager): @@ -1365,9 +1368,9 @@ def test_initialize_empty_tail_does_not_restart_message_ids(existing_session_man def test_initialize_restores_pinned_messages_after_summary_with_pin_first(existing_session_manager): """After a compaction with pin_first, restoring the agent must reattach the pinned head. - Without the fix, ``list_messages(offset=removed_message_count)`` skips the first - message — which happens to be a pinned head — so the resumed agent loses both - the pinned messages and has the offset-skim duplicate them with the summary. + ``list_messages(offset=pinned_head_count + removed_message_count)`` skips past the + still-live pinned head so the tail does not overlap with the messages reattached at + the front; the resumed agent must see exactly the live transcript in order. """ conversation_manager = SummarizingConversationManager(pin_first=3) conversation_manager.removed_message_count = 1 @@ -1397,32 +1400,6 @@ def test_initialize_restores_pinned_messages_after_summary_with_pin_first(existi assert texts == ["pinned-0", "pinned-1", "pinned-2", "summary of msg-3", "msg-4"] -def test_initialize_restores_summary_without_pin_first(existing_session_manager): - """Without pin_first, the existing behaviour of offset=removed_message_count is preserved.""" - conversation_manager = SummarizingConversationManager() - conversation_manager.removed_message_count = 1 - conversation_manager._summary_message = {"role": "user", "content": [{"text": "summary"}]} - - session_agent = SessionAgent( - agent_id="no-pin-agent", - state={}, - conversation_manager_state=conversation_manager.get_state(), - ) - existing_session_manager.session_repository.create_agent("test-session", session_agent) - - _populate_messages( - existing_session_manager.session_repository, - "no-pin-agent", - ["a", "b", "c"], - ) - - agent = Agent(agent_id="no-pin-agent", conversation_manager=SummarizingConversationManager()) - existing_session_manager.initialize(agent) - - texts = [m["content"][0]["text"] for m in agent.messages] - assert texts == ["summary", "b", "c"] - - def test_initialize_pinned_messages_no_summary_yet(existing_session_manager): """When pin_first is set but summarization hasn't run, restore must produce the full transcript and must NOT invoke the pinned-head fetch path (pinned_head_count == 0). From df2f9cb069dfd73ec4885efb2a83d4f9eff2b576 Mon Sep 17 00:00:00 2001 From: Syed Annas Date: Wed, 30 Sep 2026 14:09:57 +0100 Subject: [PATCH 7/8] fix(conversation): count every protected head message in pinned_head_count MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit is_pinned protects a pinned message's adjacent tool-pair partner as well, so trimming keeps more leading messages than pin_first names. The sliding window recorded min(pin_first, len), so with pin_first=2 and an opening tool call the live head was 3 messages but the persisted count said 2 — a restored session reattached too few, offset past the real toolResult, and the session manager fabricated an interrupted-tool result in its place. Count the protected prefix instead, and cover it with a unit test plus two FileSessionManager restore tests (plain trim and pinned tool pair). --- .../compression/pin_message.py | 18 +++ .../sliding_window_conversation_manager.py | 6 +- .../agent/test_conversation_manager.py | 25 ++++ .../test_repository_session_manager.py | 119 ++++++++++++++++++ 4 files changed, 166 insertions(+), 2 deletions(-) diff --git a/strands-py/src/strands/agent/conversation_manager/compression/pin_message.py b/strands-py/src/strands/agent/conversation_manager/compression/pin_message.py index 934a9d3c0c..ac82b14124 100644 --- a/strands-py/src/strands/agent/conversation_manager/compression/pin_message.py +++ b/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. diff --git a/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py b/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py index 6f8562e658..2ff1cc4c53 100644 --- a/strands-py/src/strands/agent/conversation_manager/sliding_window_conversation_manager.py +++ b/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__) @@ -197,7 +197,9 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A # Pin first N messages permanently (only on first reduction) if self.pin_first and self.pinned_head_count == 0: apply_pin_first(messages, self.pin_first) - self.pinned_head_count = min(self.pin_first, len(messages)) + # 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: diff --git a/strands-py/tests/strands/agent/test_conversation_manager.py b/strands-py/tests/strands/agent/test_conversation_manager.py index 8a016d053a..dac53def99 100644 --- a/strands-py/tests/strands/agent/test_conversation_manager.py +++ b/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 diff --git a/strands-py/tests/strands/session/test_repository_session_manager.py b/strands-py/tests/strands/session/test_repository_session_manager.py index d93842f785..3e4a6182b5 100644 --- a/strands-py/tests/strands/session/test_repository_session_manager.py +++ b/strands-py/tests/strands/session/test_repository_session_manager.py @@ -5,6 +5,7 @@ import pytest +from strands import tool from strands.agent.agent import Agent from strands.agent.conversation_manager.null_conversation_manager import NullConversationManager from strands.agent.conversation_manager.sliding_window_conversation_manager import SlidingWindowConversationManager @@ -1616,6 +1617,124 @@ def _converse_until_compaction(agent, model, max_turns=30): raise RuntimeError(f"{max_turns} turns produced no summarisation call") +@tool +def echo(text: str) -> str: + """Return the supplied text unchanged.""" + return text + + +class ToolCallingModel: + """A model stub that calls ``echo`` on the first turn and answers with text on every later turn.""" + + def __init__(self, context_window: int = 200): + self.context_window = context_window + self.stateful = False + self._utilization_limit_warned = False + self.calls = 0 + + def update_config(self, **model_config): + pass + + def get_config(self): + return {"context_window_limit": self.context_window} + + @property + def context_window_limit(self): + return self.context_window + + def estimate_utilization(self, input_tokens): + return input_tokens / self.context_window + + async def stream(self, messages, tool_specs=None, system_prompt=None, **kwargs): + """Emit a single ``echo`` tool call, then plain text once the call is in the transcript.""" + self.calls += 1 + called_tool = any("toolUse" in block for message in messages for block in message.get("content", [])) + + yield {"messageStart": {"role": "assistant"}} + if not called_tool: + yield {"contentBlockStart": {"start": {"toolUse": {"toolUseId": "tu1", "name": "echo"}}}} + yield {"contentBlockDelta": {"delta": {"toolUse": {"input": '{"text": "REAL RESULT"}'}}}} + yield {"contentBlockStop": {}} + yield {"messageStop": {"stopReason": "tool_use"}} + else: + yield {"contentBlockDelta": {"delta": {"text": f"answer {self.calls}"}}} + yield {"contentBlockStop": {}} + yield {"messageStop": {"stopReason": "end_turn"}} + yield { + "metadata": { + "usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2}, + "metrics": {"latencyMs": 0}, + } + } + + +def _build_sliding_file_agent(tmp_path, model, pin_first=2, window_size=4): + """Build a sliding-window agent backed by a FileSessionManager for end-to-end testing.""" + return Agent( + model=model, + system_prompt="Agent System Prompt", + tools=[echo], + conversation_manager=SlidingWindowConversationManager( + pin_first=pin_first, window_size=window_size, should_truncate_results=False + ), + session_manager=FileSessionManager(session_id="repro", storage_dir=str(tmp_path)), + callback_handler=None, + ) + + +def _transcript(agent): + """Extract role + content blocks for each message, ignoring per-message metadata such as pin markers.""" + return [(message["role"], copy.deepcopy(message["content"])) for message in agent.messages] + + +def test_file_session_sliding_window_restore_matches_in_memory(tmp_path): + """End-to-end: a resumed sliding-window session matches the live transcript after a pin_first trim.""" + agent = _build_sliding_file_agent(tmp_path, StubModel(), window_size=2) + agent("first question") + agent("second question") + agent("third question") + in_memory = _transcript(agent) + assert agent.conversation_manager.removed_message_count > 0, "no trimming happened, restore is not exercised" + + restored_agent = _build_sliding_file_agent(tmp_path, StubModel(), window_size=2) + restored = _transcript(restored_agent) + + assert restored == in_memory, ( + f"a resumed sliding-window session is not the conversation the agent had.\n" + f" in memory ({len(in_memory)}): {in_memory}\n" + f" restored ({len(restored)}): {restored}" + ) + assert restored_agent.conversation_manager.pinned_head_count == 2 + + +def test_file_session_sliding_window_restore_keeps_pinned_tool_pair(tmp_path): + """End-to-end: the restored head covers the pinned tool pair, so the real tool result survives (#4087). + + ``pin_first=2`` pins the opening user message and the assistant toolUse, and trimming also keeps the + toolResult partner. The persisted head count has to cover all three; otherwise a resumed session + reattaches a toolUse with no result and the session manager fabricates an interrupted-tool error. + """ + agent = _build_sliding_file_agent(tmp_path, ToolCallingModel()) + agent("use the tool") + agent("second question") + agent("third question") + in_memory = _transcript(agent) + assert agent.conversation_manager.pinned_head_count == 3 + assert agent.conversation_manager.removed_message_count > 0, "no trimming happened, restore is not exercised" + + restored_agent = _build_sliding_file_agent(tmp_path, ToolCallingModel()) + restored = _transcript(restored_agent) + + assert restored == in_memory, ( + f"a resumed sliding-window session lost the protected tool pair.\n" + f" in memory ({len(in_memory)}): {in_memory}\n" + f" restored ({len(restored)}): {restored}" + ) + flattened = str(restored) + assert "REAL RESULT" in flattened + assert "Tool was interrupted." not in flattened + + def test_file_session_resumed_conversation_matches_in_memory(tmp_path): """End-to-end: after compaction, rebuilding the agent on the same session restores exactly what it had. From 7947da9d6df00e792a29bdd70022f39be5516a5f Mon Sep 17 00:00:00 2001 From: Syed Annas Date: Wed, 30 Sep 2026 14:10:00 +0100 Subject: [PATCH 8/8] refactor(session): reuse pin_message and document pinned_head_count The restore path carried a private copy of pin_message; call the compression helper instead so the pin marker is written in one place. Also documents pinned_head_count on the base ConversationManager alongside removed_message_count. --- .../conversation_manager/conversation_manager.py | 3 +++ .../session/repository_session_manager.py | 16 ++-------------- 2 files changed, 5 insertions(+), 14 deletions(-) diff --git a/strands-py/src/strands/agent/conversation_manager/conversation_manager.py b/strands-py/src/strands/agent/conversation_manager/conversation_manager.py index 1e1a5ae24e..fa7d45e909 100644 --- a/strands-py/src/strands/agent/conversation_manager/conversation_manager.py +++ b/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: diff --git a/strands-py/src/strands/session/repository_session_manager.py b/strands-py/src/strands/session/repository_session_manager.py index 566228ba35..84b2349913 100644 --- a/strands-py/src/strands/session/repository_session_manager.py +++ b/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 @@ -24,19 +25,6 @@ logger = logging.getLogger(__name__) -def _pin_message_inline(messages: list[Message], index: int) -> None: - """Pin a message so it is protected from eviction during context reduction. - - Mutates the message in place by setting metadata.custom.pinned = True. - """ - message = messages[index] - metadata = message.get("metadata", {}) - custom = metadata.get("custom", {}) - custom["pinned"] = True - metadata["custom"] = custom - message["metadata"] = metadata - - class RepositorySessionManager(SessionManager[LocalAgent]): """Session manager for persisting agents in a SessionRepository. @@ -300,7 +288,7 @@ def initialize(self, agent: "LocalAgent", **kwargs: Any) -> None: 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_inline(pinned_messages, i) + pin_message(pinned_messages, i) # List the messages currently in the session, using an offset of the messages previously removed # by the conversation manager. The offset must also skip past the still-live pinned head so the