fix(anthropic): keep replay content schema-valid when every block is blank

Follow-up to the cherry-picked #68633 commits, closing the final open
review point (egilewski): _relocated_replay_cache_control was applied
only inside `if replayed:`. When anthropic_content_blocks contained
only a blank cache-marked text block, `replayed` came out empty, the
function fell through to the main path's placeholder, and the cache
marker was lost; signed thinking + a blank marked text block likewise
returned with no cacheable carrier for the relocated marker.

The replay branch now appends the non-whitespace "(empty)" placeholder
when no cacheable (text/tool_use) block survives the blank filter and a
blank text block was dropped (or a marker needs a carrier) — so replay
stays schema-valid on Bedrock/strict endpoints and the breakpoint
survives on the placeholder.

Also reconciles the block-level tests from #69517 with the new
drop-then-fallback contract (blank blocks are dropped at the block
level; the message-level result is still always non-blank).

Refs #69512

Co-authored-by: ygd58 <buraysandro9@gmail.com>

agent/anthropic_adapter.py

@@ -2020,9 +2020,12 @@ def _convert_assistant_message(m: Dict[str, Any]) -> Dict[str, Any]:
             redacted_input_by_id[_sanitize_tool_id(tc.get("id", ""))] = parsed_args
         replayed: List[Dict[str, Any]] = []
         _relocated_replay_cache_control = None
+        _dropped_blank_text = False
         for b in ordered_blocks:
             clean = _sanitize_replay_block(b)
             if clean is None:
+                if isinstance(b, dict) and b.get("type") == "text":
+                    _dropped_blank_text = True
                 if isinstance(b, dict) and isinstance(b.get("cache_control"), dict):
                     # A dropped blank text block can still carry the cache
                     # breakpoint marker -- relocate it rather than losing it.
@@ -2036,6 +2039,19 @@ def _convert_assistant_message(m: Dict[str, Any]) -> Dict[str, Any]:
                 if redacted is not None:
                     clean["input"] = redacted
             replayed.append(clean)
+        # When every text block was blank and nothing cacheable survived
+        # (e.g. signed thinking + a blank text block, or a SOLE blank
+        # cache-marked block), emit the non-whitespace placeholder so the
+        # replayed message stays schema-valid (#69512) and a relocated cache
+        # marker still has a carrier instead of being silently lost.
+        _has_cacheable_replay = any(
+            isinstance(b, dict) and b.get("type") in {"text", "tool_use"}
+            for b in replayed
+        )
+        if not _has_cacheable_replay and (
+            _dropped_blank_text or _relocated_replay_cache_control is not None
+        ):
+            replayed.append({"type": "text", "text": _EMPTY_TEXT_PLACEHOLDER})
         if replayed:
             if _relocated_replay_cache_control is not None:
                 _apply_assistant_cache_control_to_last_cacheable_block(

tests/agent/test_anthropic_adapter.py

@@ -2934,3 +2934,65 @@ class TestAllBlankFallbackAndNonStringText:
         result = self._convert(msg)  # must not raise
         blocks = result["content"]
         assert not any(b.get("type") == "text" for b in blocks)
+
+
+class TestReplayAllBlankFallback:
+    """Regression for the final open review point on #68633 (egilewski):
+
+    ``_relocated_replay_cache_control`` was applied only inside ``if
+    replayed:``. For ``anthropic_content_blocks`` containing only a blank
+    cache-marked text block, ``replayed`` became empty, the function fell
+    through to the main path's ``(empty)`` fallback, and the marker was
+    lost. A signed-thinking block plus the blank marked text also returned
+    without any relocated marker (thinking is not a cacheable carrier).
+    The replay branch now resolves a cacheable ``(empty)`` placeholder when
+    no cacheable block survives the blank filter.
+    """
+
+    def _convert(self, message):
+        from agent.anthropic_adapter import _convert_assistant_message
+        return _convert_assistant_message(message)
+
+    def test_sole_blank_marked_replay_block_keeps_marker_on_placeholder(self):
+        msg = {
+            "role": "assistant",
+            "content": "",
+            "anthropic_content_blocks": [
+                {"type": "text", "text": " ", "cache_control": {"type": "ephemeral"}},
+            ],
+        }
+        result = self._convert(msg)
+        assert result["content"] == [
+            {"type": "text", "text": "(empty)", "cache_control": {"type": "ephemeral"}}
+        ], result["content"]
+
+    def test_thinking_plus_blank_marked_text_keeps_thinking_and_marker(self):
+        msg = {
+            "role": "assistant",
+            "content": "",
+            "anthropic_content_blocks": [
+                {"type": "thinking", "thinking": "reasoning", "signature": "sig-A"},
+                {"type": "text", "text": "  ", "cache_control": {"type": "ephemeral"}},
+            ],
+        }
+        result = self._convert(msg)
+        blocks = result["content"]
+        assert blocks[0] == {"type": "thinking", "thinking": "reasoning", "signature": "sig-A"}
+        marked = [b for b in blocks if isinstance(b.get("cache_control"), dict)]
+        assert len(marked) == 1 and marked[0]["type"] == "text"
+        assert marked[0]["text"].strip(), "placeholder must be non-whitespace"
+
+    def test_thinking_plus_blank_unmarked_text_gets_schema_valid_placeholder(self):
+        """Even without a cache marker, dropping the only text block from a
+        thinking-only replay must leave schema-valid content."""
+        msg = {
+            "role": "assistant",
+            "content": "",
+            "anthropic_content_blocks": [
+                {"type": "thinking", "thinking": "reasoning", "signature": "sig-B"},
+                {"type": "text", "text": "\n"},
+            ],
+        }
+        result = self._convert(msg)
+        texts = [b for b in result["content"] if b.get("type") == "text"]
+        assert texts == [{"type": "text", "text": "(empty)"}]

tests/agent/test_anthropic_whitespace_text_blocks.py

@@ -51,13 +51,20 @@ class TestSafeText:
 
 
 class TestSanitizeReplayBlockWhitespace:
-    def test_whitespace_text_block_coerced_to_placeholder(self):
-        out = _sanitize_replay_block({"type": "text", "text": "   \n"})
-        assert out == {"type": "text", "text": _EMPTY_TEXT_PLACEHOLDER}
+    def test_whitespace_text_block_dropped(self):
+        # Blank text blocks are DROPPED at the block level (not coerced in
+        # place) — the caller relocates any cache_control the block carried
+        # and appends the non-whitespace placeholder only when nothing
+        # cacheable survives, so real thinking/tool_use blocks aren't
+        # cluttered with "(empty)" noise. See _convert_assistant_message.
+        assert _sanitize_replay_block({"type": "text", "text": "   \n"}) is None
 
-    def test_empty_text_block_coerced_to_placeholder(self):
-        out = _sanitize_replay_block({"type": "text", "text": ""})
-        assert out == {"type": "text", "text": _EMPTY_TEXT_PLACEHOLDER}
+    def test_empty_text_block_dropped(self):
+        assert _sanitize_replay_block({"type": "text", "text": ""}) is None
+
+    def test_none_text_block_dropped_without_crash(self):
+        # text=None (invalid upstream payload) must not reach .strip().
+        assert _sanitize_replay_block({"type": "text", "text": None}) is None
 
     def test_real_text_block_unchanged(self):
         out = _sanitize_replay_block({"type": "text", "text": "hi"})