1  """Tests for agent/prompt_caching.py — Anthropic cache control injection."""
  2  
  3  import copy
  4  import pytest
  5  
  6  from agent.prompt_caching import (
  7      _apply_cache_marker,
  8      apply_anthropic_cache_control,
  9  )
 10  
 11  
 12  MARKER = {"type": "ephemeral"}
 13  
 14  
 15  class TestApplyCacheMarker:
 16      def test_tool_message_gets_top_level_marker_on_native_anthropic(self):
 17          """Native Anthropic path: cache_control injected top-level (adapter moves it inside tool_result)."""
 18          msg = {"role": "tool", "content": "result"}
 19          _apply_cache_marker(msg, MARKER, native_anthropic=True)
 20          assert msg["cache_control"] == MARKER
 21  
 22      def test_tool_message_skips_marker_on_openrouter(self):
 23          """OpenRouter path: top-level cache_control on role:tool is invalid and causes silent hang."""
 24          msg = {"role": "tool", "content": "result"}
 25          _apply_cache_marker(msg, MARKER, native_anthropic=False)
 26          assert "cache_control" not in msg
 27  
 28      def test_none_content_gets_top_level_marker(self):
 29          msg = {"role": "assistant", "content": None}
 30          _apply_cache_marker(msg, MARKER)
 31          assert msg["cache_control"] == MARKER
 32  
 33      def test_empty_string_content_gets_top_level_marker(self):
 34          """Empty text blocks cannot have cache_control (Anthropic rejects them)."""
 35          msg = {"role": "assistant", "content": ""}
 36          _apply_cache_marker(msg, MARKER)
 37          assert msg["cache_control"] == MARKER
 38          # Must NOT wrap into [{"type": "text", "text": "", "cache_control": ...}]
 39          assert msg["content"] == ""
 40  
 41      def test_string_content_wrapped_in_list(self):
 42          msg = {"role": "user", "content": "Hello"}
 43          _apply_cache_marker(msg, MARKER)
 44          assert isinstance(msg["content"], list)
 45          assert len(msg["content"]) == 1
 46          assert msg["content"][0]["type"] == "text"
 47          assert msg["content"][0]["text"] == "Hello"
 48          assert msg["content"][0]["cache_control"] == MARKER
 49  
 50      def test_list_content_last_item_gets_marker(self):
 51          msg = {
 52              "role": "user",
 53              "content": [
 54                  {"type": "text", "text": "First"},
 55                  {"type": "text", "text": "Second"},
 56              ],
 57          }
 58          _apply_cache_marker(msg, MARKER)
 59          assert "cache_control" not in msg["content"][0]
 60          assert msg["content"][1]["cache_control"] == MARKER
 61  
 62      def test_empty_list_content_no_crash(self):
 63          msg = {"role": "user", "content": []}
 64          # Should not crash on empty list
 65          _apply_cache_marker(msg, MARKER)
 66  
 67  
 68  class TestApplyAnthropicCacheControl:
 69      def test_empty_messages(self):
 70          result = apply_anthropic_cache_control([])
 71          assert result == []
 72  
 73      def test_returns_deep_copy(self):
 74          msgs = [{"role": "user", "content": "Hello"}]
 75          result = apply_anthropic_cache_control(msgs)
 76          assert result is not msgs
 77          assert result[0] is not msgs[0]
 78          # Original should be unmodified
 79          assert "cache_control" not in msgs[0].get("content", "")
 80  
 81      def test_system_message_gets_marker(self):
 82          msgs = [
 83              {"role": "system", "content": "You are helpful"},
 84              {"role": "user", "content": "Hi"},
 85          ]
 86          result = apply_anthropic_cache_control(msgs)
 87          # System message should have cache_control
 88          sys_content = result[0]["content"]
 89          assert isinstance(sys_content, list)
 90          assert sys_content[0]["cache_control"]["type"] == "ephemeral"
 91  
 92      def test_last_3_non_system_get_markers(self):
 93          msgs = [
 94              {"role": "system", "content": "System"},
 95              {"role": "user", "content": "msg1"},
 96              {"role": "assistant", "content": "msg2"},
 97              {"role": "user", "content": "msg3"},
 98              {"role": "assistant", "content": "msg4"},
 99          ]
100          result = apply_anthropic_cache_control(msgs)
101          # System (index 0) + last 3 non-system (indices 2, 3, 4) = 4 breakpoints
102          # Index 1 (msg1) should NOT have marker
103          content_1 = result[1]["content"]
104          if isinstance(content_1, str):
105              assert True  # No marker applied (still a string)
106          else:
107              assert "cache_control" not in content_1[0]
108  
109      def test_no_system_message(self):
110          msgs = [
111              {"role": "user", "content": "Hello"},
112              {"role": "assistant", "content": "Hi"},
113          ]
114          result = apply_anthropic_cache_control(msgs)
115          # Both should get markers (4 slots available, only 2 messages)
116          assert len(result) == 2
117  
118      def test_1h_ttl(self):
119          msgs = [{"role": "system", "content": "System prompt"}]
120          result = apply_anthropic_cache_control(msgs, cache_ttl="1h")
121          sys_content = result[0]["content"]
122          assert isinstance(sys_content, list)
123          assert sys_content[0]["cache_control"]["ttl"] == "1h"
124  
125      def test_max_4_breakpoints(self):
126          msgs = [
127              {"role": "system", "content": "System"},
128          ] + [
129              {"role": "user" if i % 2 == 0 else "assistant", "content": f"msg{i}"}
130              for i in range(10)
131          ]
132          result = apply_anthropic_cache_control(msgs)
133          # Count how many messages have cache_control
134          count = 0
135          for msg in result:
136              content = msg.get("content")
137              if isinstance(content, list):
138                  for item in content:
139                      if isinstance(item, dict) and "cache_control" in item:
140                          count += 1
141              elif "cache_control" in msg:
142                  count += 1
143          assert count <= 4