feat(hooks): emit Before/AfterReduceContextEvent from ConversationManager

Move event emission from the single overflow call site in agent.py into
ConversationManager.reduce_context as a concrete template method. Subclasses
implement _reduce_context; the framework wraps it with Before/After event
emission so every call path (reactive overflow, proactive apply_management,
per-turn, direct tool calls) gets events automatically.

An __init_subclass__ shim detects third-party subclasses that override
reduce_context directly, transparently re-wires them to _reduce_context,
and emits a DeprecationWarning.

Closes #2048

Author: Vincent Bai

src/strands/agent/conversation_manager/conversation_manager.py

@@ -1,14 +1,19 @@
 """Abstract interface for conversation history management."""
 
+import logging
+import warnings
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Any
 
+from ...hooks.events import AfterReduceContextEvent, BeforeReduceContextEvent
 from ...hooks.registry import HookProvider, HookRegistry
 from ...types.content import Message
 
 if TYPE_CHECKING:
     from ...agent.agent import Agent
 
+logger = logging.getLogger(__name__)
+
 
 class ConversationManager(ABC, HookProvider):
     """Abstract base class for managing conversation history.
@@ -24,6 +29,10 @@ class ConversationManager(ABC, HookProvider):
     lifecycle events. Derived classes that override register_hooks must call the base implementation to ensure proper
     hook registration.
 
+    Subclasses should override ``_reduce_context`` (not ``reduce_context``) to implement their reduction strategy.
+    The framework wraps ``_reduce_context`` with ``BeforeReduceContextEvent`` / ``AfterReduceContextEvent`` emission
+    automatically.
+
     Example:
         ```python
         class MyConversationManager(ConversationManager):
@@ -33,6 +42,21 @@ def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
         ```
     """
 
+    def __init_subclass__(cls, **kwargs: Any) -> None:
+        """Detect legacy subclasses that override reduce_context directly and re-wire them."""
+        super().__init_subclass__(**kwargs)
+        if "reduce_context" in cls.__dict__ and "_reduce_context" not in cls.__dict__:
+            warnings.warn(
+                f"{cls.__name__} overrides reduce_context() directly. "
+                f"This still works but the recommended pattern is to override _reduce_context(). "
+                f"Before/AfterReduceContextEvent will continue to fire because the framework "
+                f"wraps the override transparently.",
+                DeprecationWarning,
+                stacklevel=2,
+            )
+            cls._reduce_context = cls.__dict__["reduce_context"]  # type: ignore[attr-defined]
+            del cls.reduce_context
+
     def __init__(self) -> None:
         """Initialize the ConversationManager.
 
@@ -97,12 +121,43 @@ def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         """
         pass
 
-    @abstractmethod
     def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
-        """Called when the model's context window is exceeded.
+        """Reduce the conversation context, emitting Before/After hook events automatically.
+
+        This is a concrete template method. Subclasses should override ``_reduce_context`` instead.
+
+        Args:
+            agent: The agent whose conversation history will be reduced.
+            e: The exception that triggered the context reduction, if any.
+            **kwargs: Additional keyword arguments for future extensibility.
+        """
+        message_count_before = len(agent.messages)
+        agent.hooks.invoke_callbacks(
+            BeforeReduceContextEvent(
+                agent=agent,
+                exception=e,
+                message_count=message_count_before,
+            )
+        )
+        self._reduce_context(agent, e=e, **kwargs)
+        message_count_after = len(agent.messages)
+        agent.hooks.invoke_callbacks(
+            AfterReduceContextEvent(
+                agent=agent,
+                exception=e,
+                messages_removed=message_count_before - message_count_after,
+                message_count_before=message_count_before,
+                message_count_after=message_count_after,
+            )
+        )
+
+    @abstractmethod
+    def _reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
+        """Subclass implementation of context reduction.
 
-        This method should implement the specific strategy for reducing the window size when a context overflow occurs.
-        It is typically called after a ContextWindowOverflowException is caught.
+        Called by the framework via ``reduce_context()``. Subclasses should not emit
+        ``BeforeReduceContextEvent`` / ``AfterReduceContextEvent`` themselves — the
+        framework does that automatically.
 
         Implementations might use strategies such as:
 

src/strands/agent/conversation_manager/null_conversation_manager.py

@@ -28,7 +28,7 @@ def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         """
         pass
 
-    def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
+    def _reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
         """Does not reduce context and raises an exception.
 
         Args:

src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

@@ -155,7 +155,7 @@ def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
             return
         self.reduce_context(agent)
 
-    def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
+    def _reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
         """Trim the oldest messages to reduce the conversation context size.
 
         The method handles special cases where trimming the messages leads to:

src/strands/agent/conversation_manager/summarizing_conversation_manager.py

@@ -123,7 +123,7 @@ def apply_management(self, agent: "Agent", **kwargs: Any) -> None:
         # No proactive management - summarization only happens on context overflow
         pass
 
-    def reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
+    def _reduce_context(self, agent: "Agent", e: Exception | None = None, **kwargs: Any) -> None:
         """Reduce context using summarization.
 
         Args:

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,
@@ -54,6 +56,8 @@ def log_end(self, event: AfterInvocationEvent) -> None:
     "AfterToolCallEvent",
     "BeforeModelCallEvent",
     "AfterModelCallEvent",
+    "BeforeReduceContextEvent",
+    "AfterReduceContextEvent",
     "AfterInvocationEvent",
     "MessageAddedEvent",
     "HookEvent",

src/strands/hooks/events.py

@@ -308,6 +308,58 @@ def should_reverse_callbacks(self) -> bool:
         return True
 
 
+@dataclass
+class BeforeReduceContextEvent(HookEvent):
+    """Event triggered before the agent's conversation context is reduced.
+
+    Fired whenever ``reduce_context()`` is about to run, regardless of trigger
+    (reactive overflow exception, proactive sliding-window overflow, or any
+    third-party manager's own logic).
+
+    Attributes:
+        exception: The exception that triggered context reduction, if any.
+            ``None`` when reduction was triggered proactively (e.g. sliding-window overflow).
+        message_count: The number of messages in the agent's conversation history
+            immediately before reduction is applied.
+    """
+
+    exception: Exception | None = None
+    message_count: int = 0
+
+
+@dataclass
+class AfterReduceContextEvent(HookEvent):
+    """Event triggered after the agent's conversation context has been reduced.
+
+    Fired only on successful completion of ``reduce_context()``. If the underlying
+    reduction raises, no ``AfterReduceContextEvent`` is emitted; the exception
+    propagates normally.
+
+    Note: This event uses reverse callback ordering, meaning callbacks registered
+    later will be invoked first during cleanup.
+
+    Attributes:
+        exception: The exception that triggered context reduction, if any.
+            ``None`` for proactive reductions.
+        messages_removed: Number of messages removed during this reduction
+            (``message_count_before - message_count_after``).
+        message_count_before: Number of messages in the conversation history before
+            reduction was applied.
+        message_count_after: Number of messages in the conversation history after
+            reduction completed.
+    """
+
+    exception: Exception | None = None
+    messages_removed: int = 0
+    message_count_before: int = 0
+    message_count_after: int = 0
+
+    @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
@@ -529,6 +535,78 @@ def test_agent__call__retry_with_reduced_context(mock_model, agent, tool, agener
     assert conversation_manager_spy.apply_management.call_count == 1
 
 
+def test_agent__call__emits_reduce_context_events(mock_model, agent, agenerator):
+    """Verify Before/AfterReduceContextEvent fire with correct metadata when overflow triggers reduction."""
+    messages: Messages = [
+        {"role": "user", "content": [{"text": "Hello!"}]},
+        {"role": "assistant", "content": [{"text": "Hi!"}]},
+        {"role": "user", "content": [{"text": "Whats your favorite color?"}]},
+        {"role": "assistant", "content": [{"text": "Blue!"}]},
+    ]
+    agent.messages = messages
+
+    before_events: list[BeforeReduceContextEvent] = []
+    after_events: list[AfterReduceContextEvent] = []
+
+    agent.hooks.add_callback(BeforeReduceContextEvent, before_events.append)
+    agent.hooks.add_callback(AfterReduceContextEvent, after_events.append)
+
+    trigger_exception = ContextWindowOverflowException(RuntimeError("Input is too long for requested model"))
+    mock_model.mock_stream.side_effect = [
+        trigger_exception,
+        agenerator(
+            [
+                {"contentBlockStart": {"start": {}}},
+                {"contentBlockDelta": {"delta": {"text": "Green!"}}},
+                {"contentBlockStop": {}},
+                {"messageStop": {"stopReason": "end_turn"}},
+            ]
+        ),
+    ]
+
+    agent("And now?")
+
+    assert len(before_events) == 1
+    assert len(after_events) == 1
+
+    before_event = before_events[0]
+    assert before_event.agent is agent
+    assert before_event.exception is trigger_exception
+    # Before reduction runs, the prompt "And now?" has already been appended to messages (5 total).
+    assert before_event.message_count == 5
+
+    after_event = after_events[0]
+    assert after_event.agent is agent
+    assert after_event.exception is trigger_exception
+    assert after_event.message_count_before == 5
+    assert after_event.message_count_after < after_event.message_count_before
+    assert after_event.messages_removed == after_event.message_count_before - after_event.message_count_after
+    assert after_event.messages_removed > 0
+
+
+def test_agent__call__no_reduce_context_events_on_success(mock_model, agent, agenerator):
+    """Verify reduce-context events are NOT fired on a normal successful invocation."""
+    before_events: list[BeforeReduceContextEvent] = []
+    after_events: list[AfterReduceContextEvent] = []
+
+    agent.hooks.add_callback(BeforeReduceContextEvent, before_events.append)
+    agent.hooks.add_callback(AfterReduceContextEvent, after_events.append)
+
+    mock_model.mock_stream.return_value = agenerator(
+        [
+            {"contentBlockStart": {"start": {}}},
+            {"contentBlockDelta": {"delta": {"text": "ok"}}},
+            {"contentBlockStop": {}},
+            {"messageStop": {"stopReason": "end_turn"}},
+        ]
+    )
+
+    agent("Hello?")
+
+    assert before_events == []
+    assert after_events == []
+
+
 def test_agent__call__always_sliding_window_conversation_manager_doesnt_infinite_loop(mock_model, agent, tool):
     conversation_manager = SlidingWindowConversationManager(window_size=500, should_truncate_results=False)
     conversation_manager_spy = unittest.mock.Mock(wraps=conversation_manager)