fix(skills): preserve cache points in system prompt during skills inj… (#2134)

Co-authored-by: Matthew Dai <daimatth@amazon.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: 3 people

src/strands/agent/agent.py

@@ -428,6 +428,18 @@ def system_prompt(self, value: str | list[SystemContentBlock] | None) -> None:
         """
         self._system_prompt, self._system_prompt_content = self._initialize_system_prompt(value)
 
+    @property
+    def system_prompt_content(self) -> list[SystemContentBlock] | None:
+        """Get the system prompt as a list of content blocks.
+
+        Returns the structured content block representation, preserving cache points
+        and other non-text blocks. Returns None if no system prompt is set.
+
+        Returns:
+            The system prompt as a list of content blocks, or None if no system prompt is set.
+        """
+        return list(self._system_prompt_content) if self._system_prompt_content is not None else None
+
     @property
     def tool(self) -> _ToolCaller:
         """Call tool as a function.

src/strands/vended_plugins/skills/agent_skills.py

@@ -15,6 +15,7 @@
 from ...hooks.events import BeforeInvocationEvent
 from ...plugins import Plugin, hook
 from ...tools.decorator import tool
+from ...types.content import SystemContentBlock
 from ...types.tools import ToolContext
 from .skill import Skill
 
@@ -136,34 +137,51 @@ def skills(self, skill_name: str, tool_context: ToolContext) -> str:  # noqa: D4
     def _on_before_invocation(self, event: BeforeInvocationEvent) -> None:
         """Inject skill metadata into the system prompt before each invocation.
 
-        Removes the previously injected XML block (if any) via exact string
-        replacement, then appends a fresh one. Uses agent state to track the
-        injected XML per-agent, so a single plugin instance can be shared
-        across multiple agents safely.
+        Removes the previously injected XML block (if any) via exact match,
+        then appends a fresh one. Uses agent state to track the injected XML
+        per-agent, so a single plugin instance can be shared across multiple
+        agents safely.
+
+        When the agent has a structured system prompt (list of SystemContentBlock),
+        the injection is done at the block level so that cache points and other
+        structured blocks are preserved. Otherwise falls back to string manipulation.
 
         Args:
             event: The before-invocation event containing the agent reference.
         """
         agent = event.agent
 
-        current_prompt = agent.system_prompt or ""
-
-        # Remove the previously injected XML block by exact match
         state_data = agent.state.get(self._state_key)
         last_injected_xml = state_data.get("last_injected_xml") if isinstance(state_data, dict) else None
-        if last_injected_xml is not None:
-            if last_injected_xml in current_prompt:
-                current_prompt = current_prompt.replace(last_injected_xml, "")
-            else:
-                logger.warning("unable to find previously injected skills XML in system prompt, re-appending")
 
         skills_xml = self._generate_skills_xml()
-        injection = f"\n\n{skills_xml}"
-        new_prompt = f"{current_prompt}{injection}" if current_prompt else skills_xml
-
-        new_injected_xml = injection if current_prompt else skills_xml
-        self._set_state_field(agent, "last_injected_xml", new_injected_xml)
-        agent.system_prompt = new_prompt
+        content = agent.system_prompt_content
+
+        if content is not None:
+            # Content-block path: preserve cache points and other structured blocks
+            blocks: list[SystemContentBlock] = list(content)
+            if last_injected_xml is not None:
+                injected_block: SystemContentBlock = {"text": last_injected_xml}
+                if injected_block in blocks:
+                    blocks.remove(injected_block)
+                else:
+                    logger.warning("unable to find previously injected skills XML in system prompt, re-appending")
+            blocks.append({"text": skills_xml})
+            self._set_state_field(agent, "last_injected_xml", skills_xml)
+            agent.system_prompt = blocks
+        else:
+            # String path: legacy behaviour for plain-string system prompts
+            current_prompt = agent.system_prompt or ""
+            if last_injected_xml is not None:
+                if last_injected_xml in current_prompt:
+                    current_prompt = current_prompt.replace(last_injected_xml, "")
+                else:
+                    logger.warning("unable to find previously injected skills XML in system prompt, re-appending")
+            injection = f"\n\n{skills_xml}"
+            new_prompt = f"{current_prompt}{injection}" if current_prompt else skills_xml
+            new_injected_xml = injection if current_prompt else skills_xml
+            self._set_state_field(agent, "last_injected_xml", new_injected_xml)
+            agent.system_prompt = new_prompt
 
     def get_available_skills(self) -> list[Skill]:
         """Get the list of available skills.

tests/strands/agent/test_agent.py

@@ -1166,6 +1166,33 @@ def test_system_prompt_setter_none():
     assert agent._system_prompt_content is None
 
 
+def test_system_prompt_content_string():
+    """Test that system_prompt_content returns content blocks for string prompt."""
+    agent = Agent(system_prompt="hello")
+    assert agent.system_prompt_content == [{"text": "hello"}]
+
+
+def test_system_prompt_content_structured():
+    """Test that system_prompt_content returns structured blocks with cache points."""
+    blocks = [{"text": "You are helpful"}, {"cachePoint": {"type": "default"}}]
+    agent = Agent(system_prompt=blocks)
+    assert agent.system_prompt_content == blocks
+
+
+def test_system_prompt_content_none():
+    """Test that system_prompt_content returns None when no prompt is set."""
+    agent = Agent(system_prompt=None)
+    assert agent.system_prompt_content is None
+
+
+def test_system_prompt_content_returns_copy():
+    """Test that system_prompt_content returns a defensive copy."""
+    agent = Agent(system_prompt="hello")
+    content = agent.system_prompt_content
+    content.append({"text": "injected"})
+    assert agent.system_prompt_content == [{"text": "hello"}]
+
+
 @pytest.mark.asyncio
 async def test_stream_async_passes_invocation_state(agent, mock_model, mock_event_loop_cycle, agenerator, alist):
     mock_model.mock_stream.side_effect = [

tests/strands/vended_plugins/skills/test_agent_skills.py

@@ -32,11 +32,12 @@ def _mock_agent():
     agent._system_prompt = "You are an agent."
     agent._system_prompt_content = [{"text": "You are an agent."}]
 
-    # Make system_prompt property behave like the real Agent
+    # Make system_prompt and system_prompt_content properties behave like the real Agent
     type(agent).system_prompt = property(
         lambda self: self._system_prompt,
         lambda self, value: _set_system_prompt(self, value),
     )
+    type(agent).system_prompt_content = property(lambda self: self._system_prompt_content)
 
     agent.hooks = HookRegistry()
     agent.add_hook = MagicMock(
@@ -59,11 +60,15 @@ def _mock_tool_context(agent: MagicMock) -> ToolContext:
     return ToolContext(tool_use=tool_use, agent=agent, invocation_state={"agent": agent})
 
 
-def _set_system_prompt(agent: MagicMock, value: str | None) -> None:
+def _set_system_prompt(agent: MagicMock, value: str | list | None) -> None:
     """Simulate the Agent.system_prompt setter."""
     if isinstance(value, str):
         agent._system_prompt = value
         agent._system_prompt_content = [{"text": value}]
+    elif isinstance(value, list):
+        text_parts = [block["text"] for block in value if "text" in block]
+        agent._system_prompt = "\n".join(text_parts) if text_parts else None
+        agent._system_prompt_content = value
     elif value is None:
         agent._system_prompt = None
         agent._system_prompt_content = None
@@ -417,9 +422,41 @@ def test_uses_public_system_prompt_setter(self):
         event = BeforeInvocationEvent(agent=agent)
         plugin._on_before_invocation(event)
 
-        # The public setter should have been used, so _system_prompt_content
-        # should be consistent with _system_prompt
-        assert agent._system_prompt_content == [{"text": agent._system_prompt}]
+        # The public setter should have been used via the content-block path:
+        # original block is preserved and the skills XML is appended as a new block.
+        assert len(agent.system_prompt_content) == 2
+        assert agent.system_prompt_content[0] == {"text": "Original."}
+        assert "<available_skills>" in agent.system_prompt_content[1]["text"]
+
+    def test_preserves_cache_points_in_system_prompt(self):
+        """Test that cachePoint blocks in the system prompt are preserved after injection."""
+        plugin = AgentSkills(skills=[_make_skill()])
+        agent = _mock_agent()
+        agent._system_prompt = "Base instructions."
+        agent._system_prompt_content = [
+            {"text": "Base instructions."},
+            {"cachePoint": {"type": "default"}},
+        ]
+
+        expected_skills_xml = plugin._generate_skills_xml()
+
+        event = BeforeInvocationEvent(agent=agent)
+        plugin._on_before_invocation(event)
+
+        # Exact block structure: original text, cachePoint, skills XML
+        assert agent.system_prompt_content == [
+            {"text": "Base instructions."},
+            {"cachePoint": {"type": "default"}},
+            {"text": expected_skills_xml},
+        ]
+
+        # Repeated invocation: identical result, no accumulation
+        plugin._on_before_invocation(event)
+        assert agent.system_prompt_content == [
+            {"text": "Base instructions."},
+            {"cachePoint": {"type": "default"}},
+            {"text": expected_skills_xml},
+        ]
 
     def test_warns_when_previous_xml_not_found(self, caplog):
         """Test that a warning is logged when the previously injected XML is missing from the prompt."""
@@ -441,6 +478,43 @@ def test_warns_when_previous_xml_not_found(self, caplog):
         assert "<available_skills>" in agent.system_prompt
 
 
+class TestStringPathInjection:
+    """Tests for the string-path branch of _on_before_invocation (system_prompt_content is None)."""
+
+    def test_string_path_replaces_previous_xml(self):
+        """Test that old injected XML is replaced when found in the string prompt."""
+        plugin = AgentSkills(skills=[_make_skill()])
+        agent = _mock_agent()
+
+        old_xml = "\n\n<old>xml</old>"
+        agent._system_prompt = f"Base prompt.{old_xml}"
+        agent._system_prompt_content = None
+        agent.state.set(plugin._state_key, {"last_injected_xml": old_xml})
+
+        event = BeforeInvocationEvent(agent=agent)
+        plugin._on_before_invocation(event)
+
+        assert "<old>xml</old>" not in agent.system_prompt
+        assert "<available_skills>" in agent.system_prompt
+        assert agent.system_prompt.startswith("Base prompt.")
+
+    def test_string_path_warns_when_previous_xml_not_found(self, caplog):
+        """Test that a warning is logged when old XML is missing from the string prompt."""
+        plugin = AgentSkills(skills=[_make_skill()])
+        agent = _mock_agent()
+
+        agent._system_prompt = "Totally new prompt."
+        agent._system_prompt_content = None
+        agent.state.set(plugin._state_key, {"last_injected_xml": "\n\n<old>xml</old>"})
+
+        event = BeforeInvocationEvent(agent=agent)
+        with caplog.at_level(logging.WARNING):
+            plugin._on_before_invocation(event)
+
+        assert "unable to find previously injected skills XML in system prompt" in caplog.text
+        assert "<available_skills>" in agent.system_prompt
+
+
 class TestSkillsXmlGeneration:
     """Tests for _generate_skills_xml."""