feat(hooks): add BeforeReduceContextEvent and AfterReduceContextEvent hook events

src/strands/agent/conversation_manager/null_conversation_manager.py

@@ -6,6 +6,7 @@
     from ...agent.agent import Agent
 
 from ...types.exceptions import ContextWindowOverflowException
+from ...hooks import BeforeReduceContextEvent
 from .conversation_manager import ConversationManager
 
 
@@ -40,6 +41,9 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
             e: If provided.
             ContextWindowOverflowException: If e is None.
         """
+        # Fire before event
+        agent.hooks.invoke_callbacks(BeforeReduceContextEvent(agent=agent, exception=e))
+
         if e:
             raise e
         else:

src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

@@ -6,7 +6,7 @@
 if TYPE_CHECKING:
     from ...agent.agent import Agent
 
-from ...hooks import BeforeModelCallEvent, HookRegistry
+from ...hooks import AfterReduceContextEvent, BeforeModelCallEvent, BeforeReduceContextEvent, HookRegistry
 from ...types.content import ContentBlock, Messages
 from ...types.exceptions import ContextWindowOverflowException
 from ...types.tools import ToolResultContent
@@ -171,6 +171,8 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
                 error was provided (e is not None). When called during routine window management (e is None),
                 logs a warning and returns without modification.
         """
+        agent.hooks.invoke_callbacks(BeforeReduceContextEvent(agent=agent, exception=e))
+
         messages = agent.messages
 
         # Try to truncate the tool result first
@@ -182,6 +184,7 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
             results_truncated = self._truncate_tool_results(messages, oldest_message_idx_with_tool_results)
             if results_truncated:
                 logger.debug("message_index=<%s> | tool results truncated", oldest_message_idx_with_tool_results)
+                agent.hooks.invoke_callbacks(AfterReduceContextEvent(agent=agent))
                 return
 
         # Try to trim index id when tool result cannot be truncated anymore
@@ -232,6 +235,8 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
         # Overwrite message history
         messages[:] = messages[trim_index:]
 
+        agent.hooks.invoke_callbacks(AfterReduceContextEvent(agent=agent))
+
     def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool:
         """Truncate tool results and replace image blocks in a message to reduce context size.
 

src/strands/agent/conversation_manager/summarizing_conversation_manager.py

@@ -7,6 +7,7 @@
 
 from ..._async import run_async
 from ...event_loop.streaming import process_stream
+from ...hooks import AfterReduceContextEvent, BeforeReduceContextEvent
 from ...tools._tool_helpers import noop_tool
 from ...tools.registry import ToolRegistry
 from ...types.content import Message
@@ -135,6 +136,8 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
         Raises:
             ContextWindowOverflowException: If the context cannot be summarized.
         """
+        agent.hooks.invoke_callbacks(BeforeReduceContextEvent(agent=agent, exception=e))
+
         try:
             # Calculate how many messages to summarize
             messages_to_summarize_count = max(1, int(len(agent.messages) * self.summary_ratio))
@@ -171,6 +174,8 @@ def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: A
             # Replace the summarized messages with the summary
             agent.messages[:] = [self._summary_message] + remaining_messages
 
+            agent.hooks.invoke_callbacks(AfterReduceContextEvent(agent=agent))
+
         except Exception as summarization_error:
             logger.error("Summarization failed: %s", summarization_error)
             raise summarization_error from e

src/strands/hooks/__init__.py

@@ -35,12 +35,14 @@ def log_end(self, event: AfterInvocationEvent) -> None:
     # Multiagent hook events
     AfterMultiAgentInvocationEvent,
     AfterNodeCallEvent,
+    AfterReduceContextEvent,
     AfterToolCallEvent,
     AgentInitializedEvent,
     BeforeInvocationEvent,
     BeforeModelCallEvent,
     BeforeMultiAgentInvocationEvent,
     BeforeNodeCallEvent,
+    BeforeReduceContextEvent,
     BeforeToolCallEvent,
     MessageAddedEvent,
     MultiAgentInitializedEvent,
@@ -55,6 +57,8 @@ def log_end(self, event: AfterInvocationEvent) -> None:
     "BeforeModelCallEvent",
     "AfterModelCallEvent",
     "AfterInvocationEvent",
+    "BeforeReduceContextEvent",
+    "AfterReduceContextEvent",
     "MessageAddedEvent",
     "HookEvent",
     "HookProvider",

src/strands/hooks/events.py

@@ -303,6 +303,39 @@ def should_reverse_callbacks(self) -> bool:
         return True
 
 
+@dataclass
+class BeforeReduceContextEvent(HookEvent):
+    """Event triggered before the conversation manager reduces context.
+
+    This event is fired just before the agent calls reduce_context() in response
+    to a context window overflow. Hook providers can use this event for logging,
+    observability, or displaying progress indicators during long-running sessions.
+
+    Attributes:
+        exception: The ContextWindowOverflowException that triggered the context reduction.
+    """
+
+    exception: Exception
+
+
+@dataclass
+class AfterReduceContextEvent(HookEvent):
+    """Event triggered after the conversation manager has reduced context.
+
+    This event is fired immediately after reduce_context() returns, before the
+    agent retries the model call. Hook providers can use this event to log the
+    outcome of the reduction or update observability dashboards.
+
+    Note: This event uses reverse callback ordering, meaning callbacks registered
+    later will be invoked first during cleanup.
+    """
+
+    @property
+    def should_reverse_callbacks(self) -> bool:
+        """True to invoke callbacks in reverse order."""
+        return True
+
+
 # Multiagent hook events start here
 @dataclass
 class MultiAgentInitializedEvent(BaseHookEvent):

tests/strands/agent/test_agent.py

@@ -21,7 +21,13 @@
 from strands.agent.conversation_manager.sliding_window_conversation_manager import SlidingWindowConversationManager
 from strands.agent.state import AgentState
 from strands.handlers.callback_handler import PrintingCallbackHandler, null_callback_handler
-from strands.hooks import BeforeInvocationEvent, BeforeModelCallEvent, BeforeToolCallEvent
+from strands.hooks import (
+    AfterReduceContextEvent,
+    BeforeInvocationEvent,
+    BeforeModelCallEvent,
+    BeforeReduceContextEvent,
+    BeforeToolCallEvent,
+)
 from strands.interrupt import Interrupt
 from strands.models.bedrock import DEFAULT_BEDROCK_MODEL_ID, BedrockModel
 from strands.session.repository_session_manager import RepositorySessionManager
@@ -2767,3 +2773,39 @@ def test_as_tool_defaults_description_when_agent_has_none():
     tool = agent.as_tool()
 
     assert tool.tool_spec["description"] == "Use the researcher agent as a tool by providing a natural language input"
+
+
+@pytest.mark.asyncio
+async def test_stream_async_fires_before_and_after_reduce_context_hook_events(mock_model, agent, agenerator, alist):
+    """BeforeReduceContextEvent and AfterReduceContextEvent are fired around reduce_context."""
+    overflow_exc = ContextWindowOverflowException(RuntimeError("Input is too long for requested model"))
+
+    mock_model.mock_stream.side_effect = [
+        overflow_exc,
+        agenerator(
+            [
+                {"contentBlockStart": {"contentBlockIndex": 0, "start": {}}},
+                {"contentBlockDelta": {"contentBlockIndex": 0, "delta": {"text": "OK"}}},
+                {"contentBlockStop": {"contentBlockIndex": 0}},
+                {"messageStop": {"stopReason": "end_turn"}},
+            ]
+        ),
+    ]
+
+    # Seed messages so SlidingWindowConversationManager.reduce_context() can trim them.
+    agent.messages[:] = [
+        {"role": "user", "content": [{"text": "msg1"}]},
+        {"role": "assistant", "content": [{"text": "resp1"}]},
+        {"role": "user", "content": [{"text": "msg2"}]},
+    ]
+
+    before_events = []
+    after_events = []
+    agent.add_hook(lambda e: before_events.append(e), BeforeReduceContextEvent)
+    agent.add_hook(lambda e: after_events.append(e), AfterReduceContextEvent)
+
+    await alist(agent.stream_async("hello"))
+
+    assert len(before_events) == 1
+    assert before_events[0].exception is overflow_exc
+    assert len(after_events) == 1

tests/strands/agent/test_conversation_manager.py

@@ -265,7 +265,7 @@ def test_null_conversation_manager_reduce_context_raises_context_window_overflow
     manager.apply_management(test_agent)
 
     with pytest.raises(ContextWindowOverflowException):
-        manager.reduce_context(messages)
+        manager.reduce_context(test_agent)
 
     assert messages == original_messages
 
@@ -283,7 +283,7 @@ def test_null_conversation_manager_reduce_context_with_exception_raises_same_exc
     manager.apply_management(test_agent)
 
     with pytest.raises(RuntimeError):
-        manager.reduce_context(messages, RuntimeError("test"))
+        manager.reduce_context(test_agent, RuntimeError("test"))
 
     assert messages == original_messages
 

tests/strands/agent/test_summarizing_conversation_manager.py

@@ -4,6 +4,7 @@
 import pytest
 
 from strands.agent.agent import Agent
+from strands.hooks.registry import HookRegistry
 from strands.agent.conversation_manager.summarizing_conversation_manager import (
     DEFAULT_SUMMARIZATION_PROMPT,
     SummarizingConversationManager,
@@ -45,6 +46,7 @@ def __init__(self, summary_response="This is a summary of the conversation."):
         self.summary_response = summary_response
         self.system_prompt = None
         self.messages = []
+        self.hooks = HookRegistry()
         self.model = Mock()
         self.model.stream = Mock(side_effect=lambda *a, **kw: _mock_model_stream(self.summary_response))
         self.call_tracker = Mock()