fix(event_loop): ensure tool_use content blocks are valid after max_tokens to prevent unrecoverable state

src/strands/agent/agent.py

@@ -37,7 +37,7 @@
 from ..tools.registry import ToolRegistry
 from ..tools.watcher import ToolWatcher
 from ..types.content import ContentBlock, Message, Messages
-from ..types.exceptions import ContextWindowOverflowException
+from ..types.exceptions import ContextWindowOverflowException, MaxTokensReachedException
 from ..types.tools import ToolResult, ToolUse
 from ..types.traces import AttributeValue
 from .agent_result import AgentResult
@@ -582,18 +582,21 @@ async def _execute_event_loop_cycle(self, invocation_state: dict[str, Any]) -> A
             )
             async for event in events:
                 yield event
-
+            return
         except ContextWindowOverflowException as e:
             # Try reducing the context size and retrying
-            self.conversation_manager.reduce_context(self, e=e)
+            self.conversation_manager.reduce_context(agent=self, e=e)
+        except MaxTokensReachedException as e:
+            # Recover conversation state after token limit exceeded, then continue with next cycle
+            await self.conversation_manager.handle_token_limit_reached(agent=self, e=e)
 
-            # Sync agent after reduce_context to keep conversation_manager_state up to date in the session
-            if self._session_manager:
-                self._session_manager.sync_agent(self)
+        # Sync agent after handling exception to keep conversation_manager_state up to date in the session
+        if self._session_manager:
+            self._session_manager.sync_agent(self)
 
-            events = self._execute_event_loop_cycle(invocation_state)
-            async for event in events:
-                yield event
+        events = self._execute_event_loop_cycle(invocation_state)
+        async for event in events:
+            yield event
 
     def _record_tool_execution(
         self,

src/strands/agent/conversation_manager/__init__.py

@@ -15,12 +15,14 @@
 
 from .conversation_manager import ConversationManager
 from .null_conversation_manager import NullConversationManager
+from .recover_tool_use_on_max_tokens_reached import recover_tool_use_on_max_tokens_reached
 from .sliding_window_conversation_manager import SlidingWindowConversationManager
 from .summarizing_conversation_manager import SummarizingConversationManager
 
 __all__ = [
     "ConversationManager",
     "NullConversationManager",
+    "recover_tool_use_on_max_tokens_reached",
     "SlidingWindowConversationManager",
     "SummarizingConversationManager",
 ]

src/strands/agent/conversation_manager/conversation_manager.py

@@ -4,6 +4,7 @@
 from typing import TYPE_CHECKING, Any, Optional
 
 from ...types.content import Message
+from ...types.exceptions import MaxTokensReachedException
 
 if TYPE_CHECKING:
     from ...agent.agent import Agent
@@ -86,3 +87,17 @@ def reduce_context(self, agent: "Agent", e: Optional[Exception] = None, **kwargs
             **kwargs: Additional keyword arguments for future extensibility.
         """
         pass
+
+    async def handle_token_limit_reached(self, agent: "Agent", e: MaxTokensReachedException, **kwargs: Any) -> None:
+        """Called when MaxTokensReachedException is thrown to recover conversation state.
+
+        This method should implement recovery strategies when the token limit is exceeded and the message array
+        may be in a broken state.
+
+        Args:
+            agent: The agent whose conversation state will be recovered.
+                This list is modified in-place.
+            e: The MaxTokensReachedException that triggered the recovery.
+            **kwargs: Additional keyword arguments for future extensibility.
+        """
+        raise e

src/strands/agent/conversation_manager/recover_tool_use_on_max_tokens_reached.py

@@ -0,0 +1,66 @@
+"""Shared utility for handling token limit recovery in conversation managers."""
+
+import logging
+from typing import TYPE_CHECKING
+
+from ...hooks import MessageAddedEvent
+from ...types.content import ContentBlock, Message
+from ...types.exceptions import MaxTokensReachedException
+from ...types.tools import ToolUse
+
+if TYPE_CHECKING:
+    from ...agent.agent import Agent
+
+logger = logging.getLogger(__name__)
+
+
+async def recover_tool_use_on_max_tokens_reached(agent: "Agent", exception: MaxTokensReachedException) -> None:
+    """Handle MaxTokensReachedException by cleaning up orphaned tool uses and adding corrected message.
+
+    This function fixes incomplete tool uses that may occur when the model's response is truncated
+    due to token limits. It:
+
+    1. Inspects each content block in the incomplete message for invalid tool uses
+    2. Replaces incomplete tool use blocks with informative text messages
+    3. Preserves valid content blocks in the corrected message
+    4. Adds the corrected message to the agent's conversation history
+
+    Args:
+        agent: The agent whose conversation will be updated with the corrected message.
+        exception: The MaxTokensReachedException containing the incomplete message.
+    """
+    logger.info("handling MaxTokensReachedException - inspecting incomplete message for invalid tool uses")
+
+    incomplete_message: Message = exception.incomplete_message
+
+    if not incomplete_message["content"]:
+        # Cannot correct invalid content block if content is empty
+        raise exception
+
+    valid_content: list[ContentBlock] = []
+    for content in incomplete_message["content"]:
+        tool_use: ToolUse | None = content.get("toolUse")
+        if not tool_use:
+            valid_content.append(content)
+            continue
+
+        # Check if tool use is incomplete (missing or empty required fields)
+        tool_name = tool_use.get("name")
+        if not (tool_name and tool_use.get("input") and tool_use.get("toolUseId")):
+            # Tool use is incomplete due to max_tokens truncation
+            display_name = tool_name if tool_name else "<unknown>"
+            logger.warning("tool_name=<%s> | replacing with error message due to max_tokens truncation.", display_name)
+
+            valid_content.append(
+                {
+                    "text": f"The selected tool {display_name}'s tool use was incomplete due "
+                    f"to maximum token limits being reached."
+                }
+            )
+        else:
+            # ToolUse was invalid for an unknown reason. Cannot correct, return without modifying
+            raise exception
+
+    valid_message: Message = {"content": valid_content, "role": incomplete_message["role"]}
+    agent.messages.append(valid_message)
+    agent.hooks.invoke_callbacks(MessageAddedEvent(agent=agent, message=valid_message))

src/strands/agent/conversation_manager/sliding_window_conversation_manager.py

@@ -7,8 +7,9 @@
     from ...agent.agent import Agent
 
 from ...types.content import Messages
-from ...types.exceptions import ContextWindowOverflowException
+from ...types.exceptions import ContextWindowOverflowException, MaxTokensReachedException
 from .conversation_manager import ConversationManager
+from .recover_tool_use_on_max_tokens_reached import recover_tool_use_on_max_tokens_reached
 
 logger = logging.getLogger(__name__)
 
@@ -112,6 +113,16 @@ def reduce_context(self, agent: "Agent", e: Optional[Exception] = None, **kwargs
         # Overwrite message history
         messages[:] = messages[trim_index:]
 
+    async def handle_token_limit_reached(self, agent: "Agent", e: MaxTokensReachedException, **kwargs: Any) -> None:
+        """Apply sliding window strategy for token limit recovery.
+
+        Args:
+            agent: The agent whose conversation state will be recovered.
+            e: The MaxTokensReachedException that triggered the recovery.
+            **kwargs: Additional keyword arguments for future extensibility.
+        """
+        await recover_tool_use_on_max_tokens_reached(agent, e)
+
     def _truncate_tool_results(self, messages: Messages, msg_idx: int) -> bool:
         """Truncate tool results in a message to reduce context size.
 

src/strands/agent/conversation_manager/summarizing_conversation_manager.py

@@ -6,8 +6,9 @@
 from typing_extensions import override
 
 from ...types.content import Message
-from ...types.exceptions import ContextWindowOverflowException
+from ...types.exceptions import ContextWindowOverflowException, MaxTokensReachedException
 from .conversation_manager import ConversationManager
+from .recover_tool_use_on_max_tokens_reached import recover_tool_use_on_max_tokens_reached
 
 if TYPE_CHECKING:
     from ..agent import Agent
@@ -166,6 +167,16 @@ def reduce_context(self, agent: "Agent", e: Optional[Exception] = None, **kwargs
             logger.error("Summarization failed: %s", summarization_error)
             raise summarization_error from e
 
+    async def handle_token_limit_reached(self, agent: "Agent", e: MaxTokensReachedException, **kwargs: Any) -> None:
+        """Apply summarization strategy for token limit recovery.
+
+        Args:
+            agent: The agent whose conversation state will be recovered.
+            e: The MaxTokensReachedException that triggered the recovery.
+            **kwargs: Additional keyword arguments for future extensibility.
+        """
+        await recover_tool_use_on_max_tokens_reached(agent, e)
+
     def _generate_summary(self, messages: List[Message], agent: "Agent") -> Message:
         """Generate a summary of the provided messages.
 

tests/strands/agent/conversation_manager/__init__.py

@@ -0,0 +1 @@
+# Test package for conversation manager