Add dummy toolUse message when its missing due to pagination in session management

src/strands/session/repository_session_manager.py

@@ -4,7 +4,7 @@
 from typing import TYPE_CHECKING, Any, Optional
 
 from ..agent.state import AgentState
-from ..tools._tool_helpers import generate_missing_tool_result_content
+from ..tools._tool_helpers import generate_missing_tool_result_content, generate_missing_tool_use_content
 from ..types.content import Message
 from ..types.exceptions import SessionException
 from ..types.session import (
@@ -164,12 +164,43 @@ def initialize(self, agent: "Agent", **kwargs: Any) -> None:
             agent.messages = self._fix_broken_tool_use(agent.messages)
 
     def _fix_broken_tool_use(self, messages: list[Message]) -> list[Message]:
-        """Add tool_result after orphaned tool_use messages.
+        """Fix broken tool use/result pairs in message history.
 
-        Before 1.15.0, strands had a bug where they persisted sessions with a potentially broken messages array.
-        This method retroactively fixes that issue by adding a tool_result outside of session management. After 1.15.0,
-        this bug is no longer present.
+        This method fixes two issues:
+        1. Orphaned toolUse messages without corresponding toolResult.
+           Before 1.15.0, strands had a bug where they persisted sessions with a potentially broken messages array.
+           This method retroactively fixes that issue by adding a tool_result outside of session management.
+           After 1.15.0, this bug is no longer present.
+        2. Orphaned toolResult messages without corresponding toolUse (e.g., when pagination truncates messages)
+
+        Args:
+            messages: The list of messages to fix
+            agent_id: The agent ID for fetching previous messages
+            removed_message_count: Number of messages removed by the conversation manager
+
+        Returns:
+            Fixed list of messages with proper tool use/result pairs
         """
+        # First, check if the first message has orphaned toolResult (no preceding toolUse)
+        if messages:
+            first_message = messages[0]
+            if first_message["role"] == "user" and any("toolResult" in content for content in first_message["content"]):
+                orphaned_tool_result_ids = [
+                    content["toolResult"]["toolUseId"]
+                    for content in first_message["content"]
+                    if "toolResult" in content
+                ]
+
+                if orphaned_tool_result_ids:
+                    logger.warning(
+                        "Session message history starts with orphaned toolResult(s) with no preceding toolUse. "
+                        "This typically happens when messages are truncated due to pagination limits. "
+                        "Adding dummy toolUse message to create valid conversation."
+                    )
+                    missing_tool_use_blocks = generate_missing_tool_use_content(orphaned_tool_result_ids)
+                    messages.insert(0, {"role": "assistant", "content": missing_tool_use_blocks})
+
+        # Then check for orphaned toolUse messages
         for index, message in enumerate(messages):
             # Check all but the latest message in the messages array
             # The latest message being orphaned is handled in the agent class
@@ -187,7 +218,7 @@ def _fix_broken_tool_use(self, messages: list[Message]) -> list[Message]:
                     ]
 
                     missing_tool_use_ids = list(set(tool_use_ids) - set(tool_result_ids))
-                    # If there area missing tool use ids, that means the messages history is broken
+                    # If there are missing tool use ids, that means the messages history is broken
                     if missing_tool_use_ids:
                         logger.warning(
                             "Session message history has an orphaned toolUse with no toolResult. "

src/strands/tools/_tool_helpers.py

@@ -28,3 +28,24 @@ def generate_missing_tool_result_content(tool_use_ids: list[str]) -> list[Conten
         }
         for tool_use_id in tool_use_ids
     ]
+
+
+def generate_missing_tool_use_content(tool_result_ids: list[str]) -> list[ContentBlock]:
+    """Generate ToolUse content blocks for orphaned ToolResult message.
+
+    Args:
+        tool_result_ids: List of toolUseIds from orphaned toolResult blocks
+
+    Returns:
+        List of ContentBlock dictionaries containing dummy toolUse blocks
+    """
+    return [
+        {
+            "toolUse": {
+                "toolUseId": tool_use_id,
+                "name": "unknown_tool",
+                "input": {"error": "toolUse is missing. Ignore."},
+            }
+        }
+        for tool_use_id in tool_result_ids
+    ]

tests/strands/tools/test_tool_helpers.py

@@ -0,0 +1,77 @@
+"""Tests for tool helper functions."""
+
+from strands.tools._tool_helpers import (
+    generate_missing_tool_result_content,
+    generate_missing_tool_use_content,
+)
+
+
+class TestGenerateMissingToolResultContent:
+    """Tests for generate_missing_tool_result_content function."""
+
+    def test_single_tool_use_id(self):
+        """Test generating content for a single tool use ID."""
+        tool_use_ids = ["tool_123"]
+        result = generate_missing_tool_result_content(tool_use_ids)
+
+        assert len(result) == 1
+        assert "toolResult" in result[0]
+        assert result[0]["toolResult"]["toolUseId"] == "tool_123"
+        assert result[0]["toolResult"]["status"] == "error"
+        assert result[0]["toolResult"]["content"] == [{"text": "Tool was interrupted."}]
+
+    def test_multiple_tool_use_ids(self):
+        """Test generating content for multiple tool use IDs."""
+        tool_use_ids = ["tool_123", "tool_456", "tool_789"]
+        result = generate_missing_tool_result_content(tool_use_ids)
+
+        assert len(result) == 3
+        for i, tool_id in enumerate(tool_use_ids):
+            assert "toolResult" in result[i]
+            assert result[i]["toolResult"]["toolUseId"] == tool_id
+            assert result[i]["toolResult"]["status"] == "error"
+
+    def test_empty_list(self):
+        """Test generating content for empty list."""
+        result = generate_missing_tool_result_content([])
+        assert result == []
+
+
+class TestGenerateMissingToolUseContent:
+    """Tests for generate_missing_tool_use_content function."""
+
+    def test_single_tool_result_id(self):
+        """Test generating content for a single tool result ID."""
+        tool_result_ids = ["tooluse_abc123"]
+        result = generate_missing_tool_use_content(tool_result_ids)
+
+        assert len(result) == 1
+        assert "toolUse" in result[0]
+        assert result[0]["toolUse"]["toolUseId"] == "tooluse_abc123"
+        assert result[0]["toolUse"]["name"] == "unknown_tool"
+        assert result[0]["toolUse"]["input"] == {"error": "toolUse is missing. Ignore."}
+
+    def test_multiple_tool_result_ids(self):
+        """Test generating content for multiple tool result IDs."""
+        tool_result_ids = ["tooluse_abc123", "tooluse_def456", "tooluse_ghi789"]
+        result = generate_missing_tool_use_content(tool_result_ids)
+
+        assert len(result) == 3
+        for i, tool_id in enumerate(tool_result_ids):
+            assert "toolUse" in result[i]
+            assert result[i]["toolUse"]["toolUseId"] == tool_id
+            assert result[i]["toolUse"]["name"] == "unknown_tool"
+            assert result[i]["toolUse"]["input"] == {"error": "toolUse is missing. Ignore."}
+
+    def test_empty_list(self):
+        """Test generating content for empty list."""
+        result = generate_missing_tool_use_content([])
+        assert result == []
+
+    def test_realistic_tool_use_id_format(self):
+        """Test with realistic tool use ID format (like those from Bedrock)."""
+        tool_result_ids = ["tooluse_f09Y0LwyT2yteCYshTzb_Q"]
+        result = generate_missing_tool_use_content(tool_result_ids)
+
+        assert len(result) == 1
+        assert result[0]["toolUse"]["toolUseId"] == "tooluse_f09Y0LwyT2yteCYshTzb_Q"