fix: preserve guardrail_latest_message wrapping after tool execution

src/strands/models/bedrock.py

@@ -362,6 +362,23 @@ def _inject_cache_point(self, messages: list[dict[str, Any]]) -> None:
             messages[last_assistant_idx]["content"].append({"cachePoint": {"type": "default"}})
             logger.debug("msg_idx=<%s> | added cache point to last assistant message", last_assistant_idx)
 
+    def _find_last_user_text_message_index(self, messages: Messages) -> int | None:
+        """Find the index of the last user message containing text or image content.
+
+        This is used for guardrail_latest_message to ensure that guardContent wrapping
+        targets the correct message even when toolResult messages follow.
+
+        Args:
+            messages: List of messages to search
+
+        Returns:
+            Index of the last user message with text/image content, or None if not found
+        """
+        for idx, msg in reversed(list(enumerate(messages))):
+            if msg["role"] == "user" and any("text" in cb or "image" in cb for cb in msg.get("content", [])):
+                return idx
+        return None
+
     def _format_bedrock_messages(self, messages: Messages) -> list[dict[str, Any]]:
         """Format messages for Bedrock API compatibility.
 
@@ -390,7 +407,12 @@ def _format_bedrock_messages(self, messages: Messages) -> list[dict[str, Any]]:
         filtered_unknown_members = False
         dropped_deepseek_reasoning_content = False
 
-        guardrail_latest_message = self.config.get("guardrail_latest_message", False)
+        # Pre-compute the index of the last user message containing text or image content.
+        # This ensures guardContent wrapping is maintained across tool execution cycles, where
+        # the final message in the list is a toolResult (role=user) rather than text/image content.
+        last_user_text_idx = None
+        if self.config.get("guardrail_latest_message", False):
+            last_user_text_idx = self._find_last_user_text_message_index(messages)
 
         for idx, message in enumerate(messages):
             cleaned_content: list[dict[str, Any]] = []
@@ -412,12 +434,9 @@ def _format_bedrock_messages(self, messages: Messages) -> list[dict[str, Any]]:
                 if formatted_content is None:
                     continue
 
-                # Wrap text or image content in guardrailContent if this is the last user message
-                if (
-                    guardrail_latest_message
-                    and idx == len(messages) - 1
-                    and message["role"] == "user"
-                    and ("text" in formatted_content or "image" in formatted_content)
+                # Wrap text or image content in guardContent if this is the last user text/image message
+                if idx == last_user_text_idx and (
+                    "text" in formatted_content or "image" in formatted_content
                 ):
                     if "text" in formatted_content:
                         formatted_content = {"guardContent": {"text": {"text": formatted_content["text"]}}}

tests/strands/models/test_bedrock.py

@@ -2380,6 +2380,168 @@ async def test_format_request_with_guardrail_latest_message(model):
     assert formatted_messages[2]["content"][1]["guardContent"]["image"]["format"] == "png"
 
 
+@pytest.mark.asyncio
+async def test_format_request_with_guardrail_latest_message_after_tool_use(model):
+    """Test that guardContent wraps the last user text message even when a toolResult follows it."""
+    model.update_config(
+        guardrail_id="test-guardrail",
+        guardrail_version="DRAFT",
+        guardrail_latest_message=True,
+    )
+
+    messages = [
+        {"role": "user", "content": [{"text": "First message"}]},
+        {"role": "assistant", "content": [{"text": "First response"}]},
+        {"role": "user", "content": [{"text": "what is the standard deduction?"}]},
+        {
+            "role": "assistant",
+            "content": [
+                {
+                    "toolUse": {
+                        "toolUseId": "tool-1",
+                        "name": "knowledge_base",
+                        "input": {"query": "standard deduction"},
+                    }
+                }
+            ],
+        },
+        {
+            "role": "user",
+            "content": [
+                {
+                    "toolResult": {
+                        "toolUseId": "tool-1",
+                        "content": [{"text": "The standard deduction for 2024 is $14,600."}],
+                        "status": "success",
+                    }
+                }
+            ],
+        },
+    ]
+
+    request = model._format_request(messages)
+    formatted_messages = request["messages"]
+
+    assert len(formatted_messages) == 5
+
+    # Earlier user message should NOT be wrapped
+    assert "text" in formatted_messages[0]["content"][0]
+    assert formatted_messages[0]["content"][0]["text"] == "First message"
+
+    # Last user message with text content should be wrapped, even though a toolResult comes after
+    assert "guardContent" in formatted_messages[2]["content"][0]
+    assert formatted_messages[2]["content"][0]["guardContent"]["text"]["text"] == "what is the standard deduction?"
+
+    # toolResult-only user message should NOT be wrapped
+    assert "toolResult" in formatted_messages[4]["content"][0]
+    assert "guardContent" not in formatted_messages[4]["content"][0]
+
+
+@pytest.mark.asyncio
+async def test_format_request_with_guardrail_latest_message_wraps_final_user_text(model):
+    """Test that guardContent wraps the last user message when it contains text content."""
+    model.update_config(
+        guardrail_id="test-guardrail",
+        guardrail_version="DRAFT",
+        guardrail_latest_message=True,
+    )
+
+    messages = [
+        {"role": "user", "content": [{"text": "First message"}]},
+        {"role": "assistant", "content": [{"text": "First response"}]},
+        {"role": "user", "content": [{"text": "Tell me about taxes"}]},
+    ]
+
+    request = model._format_request(messages)
+    formatted_messages = request["messages"]
+
+    assert "guardContent" in formatted_messages[2]["content"][0]
+    assert formatted_messages[2]["content"][0]["guardContent"]["text"]["text"] == "Tell me about taxes"
+
+
+@pytest.mark.asyncio
+async def test_format_request_with_guardrail_multiple_sequential_tool_calls(model):
+    """Test guardContent with multiple tool calls in sequence (no new user input between)."""
+    model.update_config(
+        guardrail_id="test-guardrail",
+        guardrail_version="DRAFT",
+        guardrail_latest_message=True,
+    )
+
+    messages = [
+        {"role": "user", "content": [{"text": "First question"}]},
+        {"role": "assistant", "content": [{"toolUse": {"toolUseId": "t1", "name": "tool1", "input": {}}}]},
+        {"role": "user", "content": [{"toolResult": {"toolUseId": "t1", "content": [{"text": "Result 1"}], "status": "success"}}]},
+        {"role": "assistant", "content": [{"toolUse": {"toolUseId": "t2", "name": "tool2", "input": {}}}]},
+        {"role": "user", "content": [{"toolResult": {"toolUseId": "t2", "content": [{"text": "Result 2"}], "status": "success"}}]},
+    ]
+
+    request = model._format_request(messages)
+    formatted_messages = request["messages"]
+
+    # Should wrap the first user text message, not the toolResults
+    assert "guardContent" in formatted_messages[0]["content"][0]
+    assert formatted_messages[0]["content"][0]["guardContent"]["text"]["text"] == "First question"
+
+    # toolResults should not be wrapped
+    assert "toolResult" in formatted_messages[2]["content"][0]
+    assert "guardContent" not in formatted_messages[2]["content"][0]
+    assert "toolResult" in formatted_messages[4]["content"][0]
+    assert "guardContent" not in formatted_messages[4]["content"][0]
+
+
+@pytest.mark.asyncio
+async def test_format_request_with_guardrail_image_before_tool_result(model):
+    """Test guardContent wraps image content even when toolResult follows."""
+    model.update_config(
+        guardrail_id="test-guardrail",
+        guardrail_version="DRAFT",
+        guardrail_latest_message=True,
+    )
+
+    messages = [
+        {"role": "user", "content": [{"image": {"format": "png", "source": {"bytes": b"fake"}}}]},
+        {"role": "assistant", "content": [{"toolUse": {"toolUseId": "t1", "name": "vision", "input": {}}}]},
+        {"role": "user", "content": [{"toolResult": {"toolUseId": "t1", "content": [{"text": "I see a cat"}], "status": "success"}}]},
+    ]
+
+    request = model._format_request(messages)
+    formatted_messages = request["messages"]
+
+    # Image should be wrapped even though toolResult comes after
+    assert "guardContent" in formatted_messages[0]["content"][0]
+    assert "image" in formatted_messages[0]["content"][0]["guardContent"]
+
+
+@pytest.mark.asyncio
+async def test_format_request_with_guardrail_multiple_tool_results_same_message(model):
+    """Test guardContent with multiple parallel tool calls (multiple toolResults in one message)."""
+    model.update_config(
+        guardrail_id="test-guardrail",
+        guardrail_version="DRAFT",
+        guardrail_latest_message=True,
+    )
+
+    messages = [
+        {"role": "user", "content": [{"text": "Question requiring multiple tools"}]},
+        {"role": "assistant", "content": [
+            {"toolUse": {"toolUseId": "t1", "name": "tool1", "input": {}}},
+            {"toolUse": {"toolUseId": "t2", "name": "tool2", "input": {}}},
+        ]},
+        {"role": "user", "content": [
+            {"toolResult": {"toolUseId": "t1", "content": [{"text": "Result 1"}], "status": "success"}},
+            {"toolResult": {"toolUseId": "t2", "content": [{"text": "Result 2"}], "status": "success"}},
+        ]},
+    ]
+
+    request = model._format_request(messages)
+    formatted_messages = request["messages"]
+
+    # Should wrap the question
+    assert "guardContent" in formatted_messages[0]["content"][0]
+    assert formatted_messages[0]["content"][0]["guardContent"]["text"]["text"] == "Question requiring multiple tools"
+
+
 def test_supports_caching_true_for_claude(bedrock_client):
     """Test that supports_caching returns True for Claude models."""
     model = BedrockModel(model_id="us.anthropic.claude-sonnet-4-20250514-v1:0")