From bd0e46449c56c830cf1e1ed8c407b5f5ec702526 Mon Sep 17 00:00:00 2001
From: raymondginger <raymondginger2018@gmail.com>
Date: Sun, 25 Oct 2026 07:59:14 +0800
Subject: [PATCH 1/2] feat(providers): anthropic prompt-cache
 clear-then-finalize + cacheHint blocks

- AnthropicProvider: strip stale cache_control markers, then place the
  single latest marker (budget 4, type ephemeral) on the newest non-system
  message before send; non-caching path strips system cache hints too.
- agent_setup: emit block-list system prompts carrying cacheHint stable/
  dynamic annotations (never leaked to any API).
- openai_compat: _sanitize_messages walks content blocks stripping
  cacheHint before request build.
- openai_responses: converter joins text blocks and strips cacheHint.

All new behavior is opt-in/annotation-driven; default flows unchanged.
---
 core/agent_setup.py                           |  27 +-
 core/providers/anthropic.py                   | 152 +++++++--
 core/providers/openai_compat.py               |  32 +-
 core/providers/openai_responses/converters.py |  17 +-
 tests/test_anthropic_cache_control.py         | 292 ++++++++++++++++++
 tests/test_cache_hint_system_blocks.py        |  80 +++++
 6 files changed, 563 insertions(+), 37 deletions(-)
 create mode 100644 tests/test_anthropic_cache_control.py
 create mode 100644 tests/test_cache_hint_system_blocks.py

diff --git a/core/agent_setup.py b/core/agent_setup.py
index f1217d6bf..39be253aa 100644
--- a/core/agent_setup.py
+++ b/core/agent_setup.py
@@ -17,8 +17,8 @@
 from __future__ import annotations
 
 import inspect
-from collections.abc import Awaitable, Callable
 import os
+from collections.abc import Awaitable, Callable
 from typing import Any
 
 from loguru import logger
@@ -370,12 +370,25 @@ def build_agent_session(
 
     if skill_runtime is None:
         skill_runtime = SkillRuntime(workspace)
-    addenda = [
-        collaboration_preamble(engine.mode),
-        system_preamble(workspace),
-    ]
-    addendum = "\n\n".join(a for a in addenda if a)
-    full_system_prompt = f"{system_prompt}\n\n{addendum}" if addendum else system_prompt
+    # Stable/dynamic split (ZCode cacheHint pattern): the base prompt plus the
+    # mode-keyed collaboration preamble rarely change, so they form the
+    # cache-stable prefix; workspace/memory sections can change between
+    # sessions and stay in a dynamic tail. When both halves exist the system
+    # prompt is emitted as two text blocks whose concatenation (the separator
+    # rides at the end of the stable block) reproduces the exact flat-join
+    # text below. cacheHint is a private key: every provider strips it before
+    # the request leaves the process (anthropic._apply_system_cache_control,
+    # openai_compat._sanitize_messages, openai_responses.converters).
+    collab_text = collaboration_preamble(engine.mode)
+    dynamic_text = system_preamble(workspace)
+    stable_text = "\n\n".join(p for p in (system_prompt, collab_text) if p)
+    if stable_text and dynamic_text:
+        full_system_prompt: str | list[dict[str, Any]] = [
+            {"type": "text", "text": f"{stable_text}\n\n", "cacheHint": "stable"},
+            {"type": "text", "text": dynamic_text, "cacheHint": "dynamic"},
+        ]
+    else:
+        full_system_prompt = stable_text or dynamic_text
 
     # Delegation: one AgentControl per top-level session drives spawn_agent /
     # wait_agent and feeds sub-agent results back through the injection seam.
diff --git a/core/providers/anthropic.py b/core/providers/anthropic.py
index 2108b2119..caf1f8120 100644
--- a/core/providers/anthropic.py
+++ b/core/providers/anthropic.py
@@ -14,9 +14,9 @@
 from core.observability import log_llm_call
 from core.providers.base import (
     LLMProvider,
+    LLMResponse,
     ProviderCapabilityError,
     ProviderConfigurationError,
-    LLMResponse,
     ReasoningDeltaCallback,
     ToolCallRequest,
 )
@@ -428,6 +428,114 @@ def _convert_tool_choice(
                 return {"type": "tool", "name": name}
         return {"type": "auto"}
 
+    @staticmethod
+    def _strip_cache_control(message: dict[str, Any]) -> dict[str, Any]:
+        """Return a copy of ``message`` with any cache_control marker removed.
+
+        Strips markers both from the message dict itself and from any content
+        block (string or block-list form) so a replayed request never carries
+        a stale breakpoint into the next finalize step.
+        """
+        if "cache_control" in message:
+            message = {k: v for k, v in message.items() if k != "cache_control"}
+        content = message.get("content")
+        if isinstance(content, list):
+            cleaned: list[Any] = []
+            changed = False
+            for block in content:
+                if isinstance(block, dict) and "cache_control" in block:
+                    cleaned.append(
+                        {k: v for k, v in block.items() if k != "cache_control"}
+                    )
+                    changed = True
+                else:
+                    cleaned.append(block)
+            if changed:
+                message = {**message, "content": cleaned}
+        return message
+
+    @staticmethod
+    def _place_cache_control(
+        message: dict[str, Any], marker: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Return a copy of ``message`` with a fresh cache_control marker.
+
+        Handles both the plain-string content shape and the block-list shape
+        (used for tool_result / multimodal content). A message with no
+        cacheable content is returned unchanged — an image-only or empty
+        result must not host a breakpoint.
+        """
+        content = message.get("content")
+        if isinstance(content, str) and content:
+            return {
+                **message,
+                "content": [{"type": "text", "text": content, "cache_control": marker}],
+            }
+        if isinstance(content, list) and content:
+            nc = list(content)
+            nc[-1] = (
+                {**nc[-1], "cache_control": marker}
+                if isinstance(nc[-1], dict)
+                else {"type": "text", "text": str(nc[-1]), "cache_control": marker}
+            )
+            return {**message, "content": nc}
+        return message
+
+    @staticmethod
+    def _strip_system_cache_hints(
+        system: str | list[dict[str, Any]],
+    ) -> str | list[dict[str, Any]]:
+        """Remove the private ``cacheHint`` section marker from system blocks.
+
+        ``cacheHint`` is an agent-side hint (see ``core.agent_setup``) telling
+        the cache layer where the stable/dynamic boundary of a block-list
+        system prompt sits. It must never reach the Anthropic API.
+        """
+        if not isinstance(system, list):
+            return system
+        if not any(isinstance(b, dict) and "cacheHint" in b for b in system):
+            return system
+        return [
+            {k: v for k, v in b.items() if k != "cacheHint"}
+            if isinstance(b, dict)
+            else b
+            for b in system
+        ]
+
+    @staticmethod
+    def _apply_system_cache_control(
+        system: str | list[dict[str, Any]], marker: dict[str, Any]
+    ) -> str | list[dict[str, Any]]:
+        """Place the single system breakpoint, honoring ``cacheHint`` sections.
+
+        The system prompt may arrive as a block list whose blocks carry a
+        private ``cacheHint`` of ``"stable"`` (base instructions + collaboration
+        preamble) or ``"dynamic"`` (project instructions, memory index). The
+        breakpoint lands on the last *stable* block so edits to the dynamic
+        tail never bust the stable-prefix cache (ZCode cacheHint pattern).
+        Without hints the marker stays on the last block (previous behavior).
+        Exactly one system breakpoint is emitted: the per-request budget is 4
+        (system 1 + tools ≤ 2 + message frontier 1). Stale markers and the
+        private hints are always stripped before send.
+        """
+        if isinstance(system, str) and system:
+            return [{"type": "text", "text": system, "cache_control": marker}]
+        if not isinstance(system, list) or not system:
+            return system
+        blocks: list[dict[str, Any]] = [
+            {k: v for k, v in block.items() if k not in ("cache_control", "cacheHint")}
+            if isinstance(block, dict)
+            else {"type": "text", "text": str(block)}
+            for block in system
+        ]
+        last_stable: int | None = None
+        for idx, block in enumerate(system):
+            if isinstance(block, dict) and block.get("cacheHint") == "stable":
+                last_stable = idx
+        place = last_stable if last_stable is not None else len(blocks) - 1
+        blocks[place] = {**blocks[place], "cache_control": marker}
+        return blocks
+
     # ------------------------------------------------------------------
     # Prompt caching
     # ------------------------------------------------------------------
@@ -443,25 +551,25 @@ def _apply_cache_control(
     ]:
         marker = {"type": "ephemeral"}
 
-        if isinstance(system, str) and system:
-            system = [{"type": "text", "text": system, "cache_control": marker}]
-        elif isinstance(system, list) and system:
-            system = list(system)
-            system[-1] = {**system[-1], "cache_control": marker}
-
-        new_msgs = list(messages)
-        if len(new_msgs) >= 3:
-            m = new_msgs[-2]
-            c = m.get("content")
-            if isinstance(c, str):
-                new_msgs[-2] = {
-                    **m,
-                    "content": [{"type": "text", "text": c, "cache_control": marker}],
-                }
-            elif isinstance(c, list) and c:
-                nc = list(c)
-                nc[-1] = {**nc[-1], "cache_control": marker}
-                new_msgs[-2] = {**m, "content": nc}
+        system = cls._apply_system_cache_control(system, marker)
+
+        # --- Message cache breakpoint: clear-then-finalize (ZCode-verified) ---
+        # The old write point hardcoded messages[-2], which silently broke when
+        # the final message was not a cacheable content block (a tool_result
+        # carrying an image, a trailing system note, a truncated turn). We now
+        # (1) strip every stale marker from the whole message list, so a reused
+        # or replayed list can never carry a leftover breakpoint that would
+        # pin the cache at the wrong offset, then (2) place a single fresh
+        # marker on the *latest non-system message*. Anchoring at the newest
+        # cacheable message (instead of a fixed -2) keeps the write point
+        # exactly at the conversation frontier, maximizing the prefix that
+        # subsequent turns hit.
+        new_msgs = [cls._strip_cache_control(m) for m in messages]
+        for idx in range(len(new_msgs) - 1, -1, -1):
+            if new_msgs[idx].get("role") == "system":
+                continue
+            new_msgs[idx] = cls._place_cache_control(new_msgs[idx], marker)
+            break
 
         new_tools = tools
         if tools:
@@ -499,6 +607,10 @@ def _build_kwargs(
                 anthropic_msgs,
                 anthropic_tools,
             )
+        else:
+            # Non-caching path: _apply_cache_control never runs, so the private
+            # cacheHint marker must be stripped here or it would leak to the API.
+            system = self._strip_system_cache_hints(system)
 
         max_tokens = max(1, max_tokens)
         effort = normalize_reasoning_effort(reasoning_effort)
diff --git a/core/providers/openai_compat.py b/core/providers/openai_compat.py
index 87254ec97..5d35244c5 100644
--- a/core/providers/openai_compat.py
+++ b/core/providers/openai_compat.py
@@ -33,9 +33,9 @@
 
 from core.providers.base import (
     LLMProvider,
+    LLMResponse,
     ProviderCapabilityError,
     ProviderConfigurationError,
-    LLMResponse,
     ReasoningDeltaCallback,
     ToolCallRequest,
 )
@@ -161,7 +161,7 @@ def _extract_tc_extras(
 
 
 def _uses_openrouter_attribution(
-    spec: "ProviderSpec | None", api_base: str | None
+    spec: ProviderSpec | None, api_base: str | None
 ) -> bool:
     """Apply Nanobot attribution headers to OpenRouter requests by default."""
     if spec and spec.name == "openrouter":
@@ -169,9 +169,7 @@ def _uses_openrouter_attribution(
     return bool(api_base and "openrouter" in api_base.lower())
 
 
-def _uses_requesty_attribution(
-    spec: "ProviderSpec | None", api_base: str | None
-) -> bool:
+def _uses_requesty_attribution(spec: ProviderSpec | None, api_base: str | None) -> bool:
     """Apply DeepCode attribution headers to Requesty requests by default."""
     if spec and spec.name == "requesty":
         return True
@@ -390,6 +388,20 @@ def map_id(value: Any) -> Any:
             return id_map.setdefault(value, self._normalize_tool_call_id(value))
 
         for clean in sanitized:
+            # cacheHint is a private agent-side key (stable/dynamic cache
+            # boundary, see core.agent_setup). It must never reach the API;
+            # message-level sanitization above only filters top-level keys, so
+            # walk content blocks here.
+            content = clean.get("content")
+            if isinstance(content, list) and any(
+                isinstance(b, dict) and "cacheHint" in b for b in content
+            ):
+                clean["content"] = [
+                    {k: v for k, v in b.items() if k != "cacheHint"}
+                    if isinstance(b, dict)
+                    else b
+                    for b in content
+                ]
             if isinstance(clean.get("tool_calls"), list):
                 normalized = []
                 for tc in clean["tool_calls"]:
@@ -416,7 +428,7 @@ def map_id(value: Any) -> Any:
                     # Some OpenAI-compatible gateways reject assistant messages
                     # that mix non-empty content with tool_calls.
                     clean["content"] = None
-            if "tool_call_id" in clean and clean["tool_call_id"]:
+            if clean.get("tool_call_id"):
                 clean["tool_call_id"] = map_id(clean["tool_call_id"])
         return self._enforce_role_alternation(sanitized)
 
@@ -545,9 +557,11 @@ def _should_use_responses_api(
 
         model_name = (model or self.default_model).lower()
         wants = False
-        if reasoning_effort and reasoning_effort.lower() != "none":
-            wants = True
-        elif any(token in model_name for token in ("gpt-5", "o1", "o3", "o4")):
+        if (
+            reasoning_effort
+            and reasoning_effort.lower() != "none"
+            or any(token in model_name for token in ("gpt-5", "o1", "o3", "o4"))
+        ):
             wants = True
         if not wants:
             return False
diff --git a/core/providers/openai_responses/converters.py b/core/providers/openai_responses/converters.py
index e0f322330..f470c6a67 100644
--- a/core/providers/openai_responses/converters.py
+++ b/core/providers/openai_responses/converters.py
@@ -26,7 +26,22 @@ def convert_messages(
         content = msg.get("content")
 
         if role == "system":
-            system_prompt = content if isinstance(content, str) else ""
+            if isinstance(content, str):
+                system_prompt = content
+            elif isinstance(content, list):
+                # Block-list system prompts (stable/dynamic cacheHint split):
+                # join the text blocks; private hint/marker keys are ignored
+                # here — the Responses API only needs the text.
+                system_prompt = "\n\n".join(
+                    block.get("text", "")
+                    for block in content
+                    if isinstance(block, dict)
+                    and block.get("type") == "text"
+                    and isinstance(block.get("text"), str)
+                    and block.get("text")
+                )
+            else:
+                system_prompt = ""
             continue
 
         if role == "user":
diff --git a/tests/test_anthropic_cache_control.py b/tests/test_anthropic_cache_control.py
new file mode 100644
index 000000000..8c5f72bc0
--- /dev/null
+++ b/tests/test_anthropic_cache_control.py
@@ -0,0 +1,292 @@
+"""Anthropic prompt caching — clear-then-finalize + cacheHint sections.
+
+Contract under test:
+
+- calling ``_apply_cache_control`` directly never raises NameError (the
+  historical bug class: an unqualified helper reference that only blew up
+  at request time on a live provider);
+- the message breakpoint is written clear-then-finalize: every stale
+  ``cache_control`` is stripped first, then ONE fresh marker lands on the
+  latest non-system message — a replayed or reused request list can never
+  carry a leftover breakpoint pinning the cache at the wrong offset;
+- the per-request marker budget is at most 4 (system 1 + tools ≤ 2 +
+  message frontier 1);
+- block-list system prompts carry a private ``cacheHint`` of stable /
+  dynamic: the breakpoint lands on the LAST STABLE block so edits to the
+  dynamic tail never bust the stable-prefix cache, and the private hint
+  key is stripped before send (it must never reach the Anthropic API);
+- a plain-string system is wrapped into a single marked block; unhinted
+  block lists keep the previous last-block behavior.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.providers.anthropic import AnthropicProvider
+
+
+def _markers(*lists: list[dict[str, Any]] | None) -> int:
+    """How many cache_control markers live across the given structures."""
+    total = 0
+    for items in lists:
+        if not items:
+            continue
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            if "cache_control" in item:
+                total += 1
+            content = item.get("content")
+            if isinstance(content, list):
+                total += sum(
+                    1 for b in content if isinstance(b, dict) and "cache_control" in b
+                )
+    return total
+
+
+def _stale_marker() -> dict[str, str]:
+    return {"type": "ephemeral"}
+
+
+def _conversation() -> list[dict[str, Any]]:
+    return [
+        {"role": "user", "content": "read the file"},
+        {
+            "role": "assistant",
+            "content": None,
+            "tool_calls": [
+                {
+                    "id": "call_1",
+                    "type": "function",
+                    "function": {"name": "read", "arguments": "{}"},
+                }
+            ],
+        },
+        {"role": "tool", "tool_call_id": "call_1", "content": "file body"},
+        {"role": "user", "content": "thanks"},
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Bare-name regression + clear-then-finalize
+# ---------------------------------------------------------------------------
+
+
+def test_apply_cache_control_is_directly_callable() -> None:
+    """The bare-name regression: classmethod-level use must not NameError."""
+    system, messages, tools = AnthropicProvider._apply_cache_control(
+        "base instructions", _conversation(), None
+    )
+    assert isinstance(system, list)
+    assert len(messages) == 4
+    assert tools is None
+
+
+def test_stale_markers_are_stripped_and_one_frontier_marker_remains() -> None:
+    messages = _conversation()
+    # A replayed list: markers on an early user message AND on the tool
+    # result, exactly the stale state the old write point could leave behind.
+    messages[0] = {
+        "role": "user",
+        "content": [
+            {"type": "text", "text": "read the file", "cache_control": _stale_marker()}
+        ],
+    }
+    messages[2] = {
+        "role": "tool",
+        "tool_call_id": "call_1",
+        "content": [
+            {"type": "text", "text": "file body", "cache_control": _stale_marker()}
+        ],
+        "cache_control": _stale_marker(),
+    }
+
+    _, out, _ = AnthropicProvider._apply_cache_control("sys", messages, None)
+
+    # Exactly one message-level breakpoint, on the LATEST non-system message.
+    marked = [
+        i
+        for i, m in enumerate(out)
+        if "cache_control" in m
+        or (
+            isinstance(m.get("content"), list)
+            and any(isinstance(b, dict) and "cache_control" in b for b in m["content"])
+        )
+    ]
+    assert marked == [3]  # final user message, string content -> wrapped block
+    # The stale message-level key on the tool result is gone.
+    assert "cache_control" not in out[2]
+    # String content at the frontier is block-wrapped, not mutated in place.
+    assert out[3]["content"] == [
+        {
+            "type": "text",
+            "text": "thanks",
+            "cache_control": {"type": "ephemeral"},
+        }
+    ]
+    assert messages[3]["content"] == "thanks"  # caller list untouched
+
+
+def test_trailing_system_note_does_not_steal_the_frontier() -> None:
+    messages = _conversation() + [{"role": "system", "content": "[note]"}]
+    _, out, _ = AnthropicProvider._apply_cache_control("sys", messages, None)
+    # Frontier marker sits on the last NON-system message, not the note.
+    # String content is block-wrapped, so check both message-level keys and
+    # content-block markers.
+    marked = [
+        i
+        for i, m in enumerate(out)
+        if "cache_control" in m
+        or (
+            isinstance(m.get("content"), list)
+            and any(isinstance(b, dict) and "cache_control" in b for b in m["content"])
+        )
+    ]
+    assert marked == [3]
+
+
+def test_all_system_history_places_no_message_marker() -> None:
+    messages = [{"role": "system", "content": "only notes"}]
+    _, out, _ = AnthropicProvider._apply_cache_control("sys", messages, None)
+    assert _markers(out) == 0
+
+
+# ---------------------------------------------------------------------------
+# Marker budget
+# ---------------------------------------------------------------------------
+
+
+def test_marker_budget_never_exceeds_four() -> None:
+    tools = [
+        {"name": "read", "description": "", "input_schema": {}},
+        {"name": "bash", "description": "", "input_schema": {}},
+        {"name": "edit", "description": "", "input_schema": {}},
+        {"name": "mcp__github__createIssue", "description": "", "input_schema": {}},
+        {"name": "mcp__github__listIssues", "description": "", "input_schema": {}},
+    ]
+    system, out, out_tools = AnthropicProvider._apply_cache_control(
+        "base instructions", _conversation(), tools
+    )
+    total = _markers(system if isinstance(system, list) else None, out, out_tools)
+    assert total <= 4
+    assert total >= 3  # system 1 + message 1 + at least one tool
+
+
+def test_tool_markers_cover_last_builtin_and_tail() -> None:
+    tools = [
+        {"name": "read", "description": "", "input_schema": {}},
+        {"name": "bash", "description": "", "input_schema": {}},
+        {"name": "mcp__ghidra__decompile", "description": "", "input_schema": {}},
+    ]
+    _, _, out_tools = AnthropicProvider._apply_cache_control(
+        "sys", _conversation(), tools
+    )
+    marked = [i for i, t in enumerate(out_tools or []) if "cache_control" in t]
+    # Last builtin (bash, index 1) and the MCP tail (index 2), never more.
+    assert marked == [1, 2]
+
+
+def test_empty_or_none_tools_pass_through() -> None:
+    _, _, no_tools = AnthropicProvider._apply_cache_control(
+        "sys", _conversation(), None
+    )
+    _, _, empty_tools = AnthropicProvider._apply_cache_control(
+        "sys", _conversation(), []
+    )
+    assert no_tools is None
+    assert empty_tools == []
+
+
+# ---------------------------------------------------------------------------
+# System breakpoint + cacheHint sections
+# ---------------------------------------------------------------------------
+
+
+def test_string_system_becomes_one_marked_block() -> None:
+    system, _, _ = AnthropicProvider._apply_cache_control(
+        "base instructions", _conversation(), None
+    )
+    assert system == [
+        {
+            "type": "text",
+            "text": "base instructions",
+            "cache_control": {"type": "ephemeral"},
+        }
+    ]
+
+
+def test_cache_hint_stable_boundary_wins_and_hints_are_stripped() -> None:
+    system_blocks: list[dict[str, Any]] = [
+        {"type": "text", "text": "persona", "cacheHint": "stable"},
+        {"type": "text", "text": "collab preamble", "cacheHint": "stable"},
+        {"type": "text", "text": "project docs", "cacheHint": "dynamic"},
+        {"type": "text", "text": "memory index", "cacheHint": "dynamic"},
+    ]
+    system, _, _ = AnthropicProvider._apply_cache_control(
+        system_blocks, _conversation(), None
+    )
+    assert isinstance(system, list)
+    # Breakpoint on the LAST STABLE block (index 1), not the list tail.
+    marked = [i for i, b in enumerate(system) if "cache_control" in b]
+    assert marked == [1]
+    # The private hint key never survives to the API payload.
+    assert all("cacheHint" not in b for b in system)
+    # Block order and text are preserved.
+    assert [b["text"] for b in system] == [
+        "persona",
+        "collab preamble",
+        "project docs",
+        "memory index",
+    ]
+
+
+def test_unhinted_block_list_keeps_last_block_behavior() -> None:
+    system_blocks: list[dict[str, Any]] = [
+        {"type": "text", "text": "a"},
+        {"type": "text", "text": "b"},
+    ]
+    system, _, _ = AnthropicProvider._apply_cache_control(
+        system_blocks, _conversation(), None
+    )
+    assert isinstance(system, list)
+    assert [i for i, b in enumerate(system) if "cache_control" in b] == [1]
+
+
+def test_stale_system_markers_are_stripped_before_finalize() -> None:
+    system_blocks: list[dict[str, Any]] = [
+        {"type": "text", "text": "a", "cache_control": _stale_marker()},
+        {"type": "text", "text": "b", "cache_control": _stale_marker()},
+    ]
+    system, _, _ = AnthropicProvider._apply_cache_control(
+        system_blocks, _conversation(), None
+    )
+    assert isinstance(system, list)
+    assert [i for i, b in enumerate(system) if "cache_control" in b] == [1]
+
+
+def test_empty_system_passes_through_unchanged() -> None:
+    empty_list, _, _ = AnthropicProvider._apply_cache_control([], _conversation(), None)
+    empty_str, _, _ = AnthropicProvider._apply_cache_control("", _conversation(), None)
+    assert empty_list == []
+    assert empty_str == ""
+
+
+def test_strip_system_cache_hints_is_copy_on_write() -> None:
+    blocks = [
+        {"type": "text", "text": "a", "cacheHint": "stable"},
+        {"type": "text", "text": "b", "cacheHint": "dynamic"},
+    ]
+    stripped = AnthropicProvider._strip_system_cache_hints(blocks)
+    assert all("cacheHint" not in b for b in stripped)
+    # Input untouched (copy-on-write): the caller's history keeps its hint.
+    assert all("cacheHint" in b for b in blocks)
+    # Idempotent + str passthrough.
+    assert AnthropicProvider._strip_system_cache_hints(stripped) == stripped
+    assert AnthropicProvider._strip_system_cache_hints("plain") == "plain"
diff --git a/tests/test_cache_hint_system_blocks.py b/tests/test_cache_hint_system_blocks.py
new file mode 100644
index 000000000..16ed126be
--- /dev/null
+++ b/tests/test_cache_hint_system_blocks.py
@@ -0,0 +1,80 @@
+"""cacheHint hygiene on the non-Anthropic paths.
+
+``cacheHint`` is a private agent-side key (see ``core.agent_setup``) that
+marks the stable/dynamic boundary of a block-list system prompt for the
+Anthropic cache layer. Providers that do not understand it must strip it
+before the payload leaves the process, and must not lose the text it
+annotates.
+
+Contract under test:
+
+- OpenAI-compat message sanitization removes ``cacheHint`` from content
+  BLOCKS (top-level key filtering alone would leak it);
+- the Responses-API converter joins block-list system prompts into plain
+  text, ignoring ``cacheHint`` entirely — the hint key is Anthropic-only
+  plumbing and must not appear anywhere in the converted payload.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.providers.openai_compat import OpenAICompatProvider
+from core.providers.openai_responses.converters import convert_messages
+
+
+def _hinted_system() -> list[dict[str, Any]]:
+    return [
+        {"type": "text", "text": "persona", "cacheHint": "stable"},
+        {"type": "text", "text": "project docs", "cacheHint": "dynamic"},
+    ]
+
+
+def test_openai_compat_sanitizer_strips_cache_hint_from_blocks() -> None:
+    provider = OpenAICompatProvider(api_key="test-key")
+    messages: list[dict[str, Any]] = [
+        {"role": "system", "content": _hinted_system()},
+        {"role": "user", "content": "hello"},
+    ]
+    sanitized = provider._sanitize_messages(messages)
+
+    system_content = sanitized[0]["content"]
+    assert isinstance(system_content, list)
+    assert all("cacheHint" not in block for block in system_content)
+    # The annotated text survives — only the private key is removed.
+    assert [block["text"] for block in system_content] == ["persona", "project docs"]
+    # The original caller list is untouched (sanitization is copy-on-write).
+    assert all("cacheHint" in block for block in messages[0]["content"])
+
+
+def test_responses_converter_joins_hinted_blocks_and_drops_the_key() -> None:
+    messages: list[dict[str, Any]] = [
+        {"role": "system", "content": _hinted_system()},
+        {"role": "user", "content": "hello"},
+    ]
+    system_prompt, input_items = convert_messages(messages)
+
+    # Hint keys are Anthropic-only plumbing: the joined text is clean.
+    assert system_prompt == "persona\n\nproject docs"
+    assert "cacheHint" not in system_prompt
+    assert "cacheHint" not in str(input_items)
+    # The converter wraps string content into Responses-API input_text blocks.
+    assert input_items == [
+        {"role": "user", "content": [{"type": "input_text", "text": "hello"}]}
+    ]
+
+
+def test_responses_converter_string_system_passthrough() -> None:
+    system_prompt, input_items = convert_messages(
+        [{"role": "system", "content": "plain"}, {"role": "user", "content": "hi"}]
+    )
+    assert system_prompt == "plain"
+    assert input_items == [
+        {"role": "user", "content": [{"type": "input_text", "text": "hi"}]}
+    ]

From 68dfed8750c1732a72f37dd3e06e35eddbce6e19 Mon Sep 17 00:00:00 2001
From: raymondginger <raymondginger2018@gmail.com>
Date: Tue, 27 Oct 2026 21:58:18 +0800
Subject: [PATCH 2/2] refactor(providers): drop cacheHint blocks, keep the
 frontier breakpoint

Address review on PR #280:

- The stable/dynamic cacheHint split is gone: the system prompt stays
  flat again, and the marker-stripping hooks in the OpenAI-compatible
  provider and the Responses converters are removed with it.
- The anthropic clear-then-finalize change stays: strip every stale
  marker from the replayed list, then place one fresh marker on the
  latest non-system message instead of a hardcoded messages[-2].
- The cache tests now drive _build_kwargs with converted messages
  instead of reaching into _prepare_request_kwargs, and the legacy
  cacheHint tests are deleted.
---
 core/agent_setup.py                           |  27 +-
 core/providers/anthropic.py                   |  77 +----
 core/providers/openai_compat.py               |  32 +-
 core/providers/openai_responses/converters.py |  17 +-
 tests/test_anthropic_cache_control.py         | 316 +++++++-----------
 tests/test_cache_hint_system_blocks.py        |  80 -----
 6 files changed, 157 insertions(+), 392 deletions(-)
 delete mode 100644 tests/test_cache_hint_system_blocks.py

diff --git a/core/agent_setup.py b/core/agent_setup.py
index 39be253aa..f1217d6bf 100644
--- a/core/agent_setup.py
+++ b/core/agent_setup.py
@@ -17,8 +17,8 @@
 from __future__ import annotations
 
 import inspect
-import os
 from collections.abc import Awaitable, Callable
+import os
 from typing import Any
 
 from loguru import logger
@@ -370,25 +370,12 @@ def build_agent_session(
 
     if skill_runtime is None:
         skill_runtime = SkillRuntime(workspace)
-    # Stable/dynamic split (ZCode cacheHint pattern): the base prompt plus the
-    # mode-keyed collaboration preamble rarely change, so they form the
-    # cache-stable prefix; workspace/memory sections can change between
-    # sessions and stay in a dynamic tail. When both halves exist the system
-    # prompt is emitted as two text blocks whose concatenation (the separator
-    # rides at the end of the stable block) reproduces the exact flat-join
-    # text below. cacheHint is a private key: every provider strips it before
-    # the request leaves the process (anthropic._apply_system_cache_control,
-    # openai_compat._sanitize_messages, openai_responses.converters).
-    collab_text = collaboration_preamble(engine.mode)
-    dynamic_text = system_preamble(workspace)
-    stable_text = "\n\n".join(p for p in (system_prompt, collab_text) if p)
-    if stable_text and dynamic_text:
-        full_system_prompt: str | list[dict[str, Any]] = [
-            {"type": "text", "text": f"{stable_text}\n\n", "cacheHint": "stable"},
-            {"type": "text", "text": dynamic_text, "cacheHint": "dynamic"},
-        ]
-    else:
-        full_system_prompt = stable_text or dynamic_text
+    addenda = [
+        collaboration_preamble(engine.mode),
+        system_preamble(workspace),
+    ]
+    addendum = "\n\n".join(a for a in addenda if a)
+    full_system_prompt = f"{system_prompt}\n\n{addendum}" if addendum else system_prompt
 
     # Delegation: one AgentControl per top-level session drives spawn_agent /
     # wait_agent and feeds sub-agent results back through the injection seam.
diff --git a/core/providers/anthropic.py b/core/providers/anthropic.py
index caf1f8120..0feb29ee2 100644
--- a/core/providers/anthropic.py
+++ b/core/providers/anthropic.py
@@ -14,9 +14,9 @@
 from core.observability import log_llm_call
 from core.providers.base import (
     LLMProvider,
-    LLMResponse,
     ProviderCapabilityError,
     ProviderConfigurationError,
+    LLMResponse,
     ReasoningDeltaCallback,
     ToolCallRequest,
 )
@@ -481,61 +481,6 @@ def _place_cache_control(
             return {**message, "content": nc}
         return message
 
-    @staticmethod
-    def _strip_system_cache_hints(
-        system: str | list[dict[str, Any]],
-    ) -> str | list[dict[str, Any]]:
-        """Remove the private ``cacheHint`` section marker from system blocks.
-
-        ``cacheHint`` is an agent-side hint (see ``core.agent_setup``) telling
-        the cache layer where the stable/dynamic boundary of a block-list
-        system prompt sits. It must never reach the Anthropic API.
-        """
-        if not isinstance(system, list):
-            return system
-        if not any(isinstance(b, dict) and "cacheHint" in b for b in system):
-            return system
-        return [
-            {k: v for k, v in b.items() if k != "cacheHint"}
-            if isinstance(b, dict)
-            else b
-            for b in system
-        ]
-
-    @staticmethod
-    def _apply_system_cache_control(
-        system: str | list[dict[str, Any]], marker: dict[str, Any]
-    ) -> str | list[dict[str, Any]]:
-        """Place the single system breakpoint, honoring ``cacheHint`` sections.
-
-        The system prompt may arrive as a block list whose blocks carry a
-        private ``cacheHint`` of ``"stable"`` (base instructions + collaboration
-        preamble) or ``"dynamic"`` (project instructions, memory index). The
-        breakpoint lands on the last *stable* block so edits to the dynamic
-        tail never bust the stable-prefix cache (ZCode cacheHint pattern).
-        Without hints the marker stays on the last block (previous behavior).
-        Exactly one system breakpoint is emitted: the per-request budget is 4
-        (system 1 + tools ≤ 2 + message frontier 1). Stale markers and the
-        private hints are always stripped before send.
-        """
-        if isinstance(system, str) and system:
-            return [{"type": "text", "text": system, "cache_control": marker}]
-        if not isinstance(system, list) or not system:
-            return system
-        blocks: list[dict[str, Any]] = [
-            {k: v for k, v in block.items() if k not in ("cache_control", "cacheHint")}
-            if isinstance(block, dict)
-            else {"type": "text", "text": str(block)}
-            for block in system
-        ]
-        last_stable: int | None = None
-        for idx, block in enumerate(system):
-            if isinstance(block, dict) and block.get("cacheHint") == "stable":
-                last_stable = idx
-        place = last_stable if last_stable is not None else len(blocks) - 1
-        blocks[place] = {**blocks[place], "cache_control": marker}
-        return blocks
-
     # ------------------------------------------------------------------
     # Prompt caching
     # ------------------------------------------------------------------
@@ -551,12 +496,16 @@ def _apply_cache_control(
     ]:
         marker = {"type": "ephemeral"}
 
-        system = cls._apply_system_cache_control(system, marker)
-
-        # --- Message cache breakpoint: clear-then-finalize (ZCode-verified) ---
-        # The old write point hardcoded messages[-2], which silently broke when
-        # the final message was not a cacheable content block (a tool_result
-        # carrying an image, a trailing system note, a truncated turn). We now
+        if isinstance(system, str) and system:
+            system = [{"type": "text", "text": system, "cache_control": marker}]
+        elif isinstance(system, list) and system:
+            system = list(system)
+            system[-1] = {**system[-1], "cache_control": marker}
+
+        # --- Message cache breakpoint: clear-then-finalize ---
+        # The previous write point hardcoded messages[-2], which broke when the
+        # final message was not a cacheable content block (a tool_result
+        # carrying an image, a trailing system note, a truncated turn). Now
         # (1) strip every stale marker from the whole message list, so a reused
         # or replayed list can never carry a leftover breakpoint that would
         # pin the cache at the wrong offset, then (2) place a single fresh
@@ -607,10 +556,6 @@ def _build_kwargs(
                 anthropic_msgs,
                 anthropic_tools,
             )
-        else:
-            # Non-caching path: _apply_cache_control never runs, so the private
-            # cacheHint marker must be stripped here or it would leak to the API.
-            system = self._strip_system_cache_hints(system)
 
         max_tokens = max(1, max_tokens)
         effort = normalize_reasoning_effort(reasoning_effort)
diff --git a/core/providers/openai_compat.py b/core/providers/openai_compat.py
index 5d35244c5..87254ec97 100644
--- a/core/providers/openai_compat.py
+++ b/core/providers/openai_compat.py
@@ -33,9 +33,9 @@
 
 from core.providers.base import (
     LLMProvider,
-    LLMResponse,
     ProviderCapabilityError,
     ProviderConfigurationError,
+    LLMResponse,
     ReasoningDeltaCallback,
     ToolCallRequest,
 )
@@ -161,7 +161,7 @@ def _extract_tc_extras(
 
 
 def _uses_openrouter_attribution(
-    spec: ProviderSpec | None, api_base: str | None
+    spec: "ProviderSpec | None", api_base: str | None
 ) -> bool:
     """Apply Nanobot attribution headers to OpenRouter requests by default."""
     if spec and spec.name == "openrouter":
@@ -169,7 +169,9 @@ def _uses_openrouter_attribution(
     return bool(api_base and "openrouter" in api_base.lower())
 
 
-def _uses_requesty_attribution(spec: ProviderSpec | None, api_base: str | None) -> bool:
+def _uses_requesty_attribution(
+    spec: "ProviderSpec | None", api_base: str | None
+) -> bool:
     """Apply DeepCode attribution headers to Requesty requests by default."""
     if spec and spec.name == "requesty":
         return True
@@ -388,20 +390,6 @@ def map_id(value: Any) -> Any:
             return id_map.setdefault(value, self._normalize_tool_call_id(value))
 
         for clean in sanitized:
-            # cacheHint is a private agent-side key (stable/dynamic cache
-            # boundary, see core.agent_setup). It must never reach the API;
-            # message-level sanitization above only filters top-level keys, so
-            # walk content blocks here.
-            content = clean.get("content")
-            if isinstance(content, list) and any(
-                isinstance(b, dict) and "cacheHint" in b for b in content
-            ):
-                clean["content"] = [
-                    {k: v for k, v in b.items() if k != "cacheHint"}
-                    if isinstance(b, dict)
-                    else b
-                    for b in content
-                ]
             if isinstance(clean.get("tool_calls"), list):
                 normalized = []
                 for tc in clean["tool_calls"]:
@@ -428,7 +416,7 @@ def map_id(value: Any) -> Any:
                     # Some OpenAI-compatible gateways reject assistant messages
                     # that mix non-empty content with tool_calls.
                     clean["content"] = None
-            if clean.get("tool_call_id"):
+            if "tool_call_id" in clean and clean["tool_call_id"]:
                 clean["tool_call_id"] = map_id(clean["tool_call_id"])
         return self._enforce_role_alternation(sanitized)
 
@@ -557,11 +545,9 @@ def _should_use_responses_api(
 
         model_name = (model or self.default_model).lower()
         wants = False
-        if (
-            reasoning_effort
-            and reasoning_effort.lower() != "none"
-            or any(token in model_name for token in ("gpt-5", "o1", "o3", "o4"))
-        ):
+        if reasoning_effort and reasoning_effort.lower() != "none":
+            wants = True
+        elif any(token in model_name for token in ("gpt-5", "o1", "o3", "o4")):
             wants = True
         if not wants:
             return False
diff --git a/core/providers/openai_responses/converters.py b/core/providers/openai_responses/converters.py
index f470c6a67..e0f322330 100644
--- a/core/providers/openai_responses/converters.py
+++ b/core/providers/openai_responses/converters.py
@@ -26,22 +26,7 @@ def convert_messages(
         content = msg.get("content")
 
         if role == "system":
-            if isinstance(content, str):
-                system_prompt = content
-            elif isinstance(content, list):
-                # Block-list system prompts (stable/dynamic cacheHint split):
-                # join the text blocks; private hint/marker keys are ignored
-                # here — the Responses API only needs the text.
-                system_prompt = "\n\n".join(
-                    block.get("text", "")
-                    for block in content
-                    if isinstance(block, dict)
-                    and block.get("type") == "text"
-                    and isinstance(block.get("text"), str)
-                    and block.get("text")
-                )
-            else:
-                system_prompt = ""
+            system_prompt = content if isinstance(content, str) else ""
             continue
 
         if role == "user":
diff --git a/tests/test_anthropic_cache_control.py b/tests/test_anthropic_cache_control.py
index 8c5f72bc0..81fe45716 100644
--- a/tests/test_anthropic_cache_control.py
+++ b/tests/test_anthropic_cache_control.py
@@ -1,22 +1,22 @@
-"""Anthropic prompt caching — clear-then-finalize + cacheHint sections.
+"""Anthropic prompt caching — clear-then-finalize at the conversation frontier.
+
+Every test drives ``AnthropicProvider._build_kwargs`` so the input messages go
+through the provider's own conversion exactly like a live request; assertions
+run on the converted Anthropic-format payload inside ``kwargs``.
 
 Contract under test:
 
-- calling ``_apply_cache_control`` directly never raises NameError (the
-  historical bug class: an unqualified helper reference that only blew up
-  at request time on a live provider);
 - the message breakpoint is written clear-then-finalize: every stale
-  ``cache_control`` is stripped first, then ONE fresh marker lands on the
-  latest non-system message — a replayed or reused request list can never
-  carry a leftover breakpoint pinning the cache at the wrong offset;
-- the per-request marker budget is at most 4 (system 1 + tools ≤ 2 +
-  message frontier 1);
-- block-list system prompts carry a private ``cacheHint`` of stable /
-  dynamic: the breakpoint lands on the LAST STABLE block so edits to the
-  dynamic tail never bust the stable-prefix cache, and the private hint
-  key is stripped before send (it must never reach the Anthropic API);
-- a plain-string system is wrapped into a single marked block; unhinted
-  block lists keep the previous last-block behavior.
+  ``cache_control`` is stripped from the converted list, then ONE fresh
+  marker lands on the LATEST message — the previous fixed ``messages[-2]``
+  write point missed the frontier whenever the tail shape changed;
+- a trailing system note never steals the frontier (it folds into the system
+  prompt, and the marker stays on the last real message);
+- the per-request marker budget is at most 4 (system 1 + tools <= 2 +
+  message frontier 1), with tool markers on the last builtin and the MCP tail;
+- a string system prompt is wrapped into exactly one marked block;
+- ``supports_caching=False`` emits no markers at all, and empty tool lists
+  never produce a ``tools`` kwarg.
 """
 
 from __future__ import annotations
@@ -32,10 +32,50 @@
 from core.providers.anthropic import AnthropicProvider
 
 
-def _markers(*lists: list[dict[str, Any]] | None) -> int:
+def _provider() -> AnthropicProvider:
+    provider = AnthropicProvider(api_key="test-key", default_model="claude-sonnet-4-6")
+    provider.reasoning_supported = False
+    return provider
+
+
+def _build(
+    messages: list[dict[str, Any]],
+    tools: list[dict[str, Any]] | None = None,
+    *,
+    supports_caching: bool = True,
+) -> dict[str, Any]:
+    """Run the real kwargs path: OpenAI-format in, Anthropic-format payload out."""
+    return _provider()._build_kwargs(
+        messages,
+        tools,
+        "claude-sonnet-4-6",
+        1024,
+        0.0,
+        None,
+        None,
+        supports_caching=supports_caching,
+    )
+
+
+def _marked_indices(messages: list[dict[str, Any]]) -> list[int]:
+    """Indices of messages carrying a cache_control marker (key or block)."""
+    marked: list[int] = []
+    for i, msg in enumerate(messages):
+        if "cache_control" in msg:
+            marked.append(i)
+            continue
+        content = msg.get("content")
+        if isinstance(content, list) and any(
+            isinstance(b, dict) and "cache_control" in b for b in content
+        ):
+            marked.append(i)
+    return marked
+
+
+def _count_markers(*structures: Any) -> int:
     """How many cache_control markers live across the given structures."""
     total = 0
-    for items in lists:
+    for items in structures:
         if not items:
             continue
         for item in items:
@@ -51,10 +91,6 @@ def _markers(*lists: list[dict[str, Any]] | None) -> int:
     return total
 
 
-def _stale_marker() -> dict[str, str]:
-    return {"type": "ephemeral"}
-
-
 def _conversation() -> list[dict[str, Any]]:
     return [
         {"role": "user", "content": "read the file"},
@@ -74,88 +110,60 @@ def _conversation() -> list[dict[str, Any]]:
     ]
 
 
+def _tools(*names: str) -> list[dict[str, Any]]:
+    return [
+        {"type": "function", "function": {"name": name, "parameters": {}}}
+        for name in names
+    ]
+
+
 # ---------------------------------------------------------------------------
-# Bare-name regression + clear-then-finalize
+# Frontier placement (the change under review)
 # ---------------------------------------------------------------------------
 
 
-def test_apply_cache_control_is_directly_callable() -> None:
-    """The bare-name regression: classmethod-level use must not NameError."""
-    system, messages, tools = AnthropicProvider._apply_cache_control(
-        "base instructions", _conversation(), None
-    )
-    assert isinstance(system, list)
-    assert len(messages) == 4
-    assert tools is None
+def test_frontier_marker_lands_on_latest_message() -> None:
+    kwargs = _build(_conversation())
+    out = kwargs["messages"]
+    assert len(out) >= 3
+    # ONE message-level breakpoint, on the LATEST message — not the fixed
+    # messages[-2] write point, which marked the tool result instead.
+    assert _marked_indices(out) == [len(out) - 1]
+
+
+def test_trailing_system_note_does_not_steal_the_frontier() -> None:
+    kwargs = _build(_conversation() + [{"role": "system", "content": "[note]"}])
+    out = kwargs["messages"]
+    # The note folds into the system prompt; the frontier marker stays on the
+    # last real conversation message.
+    assert "[note]" in str(kwargs.get("system"))
+    assert _marked_indices(out) == [len(out) - 1]
 
 
 def test_stale_markers_are_stripped_and_one_frontier_marker_remains() -> None:
     messages = _conversation()
-    # A replayed list: markers on an early user message AND on the tool
-    # result, exactly the stale state the old write point could leave behind.
+    # A replayed list: a stale marker on an early user message, exactly the
+    # leftover state the old write point could never clean up.
     messages[0] = {
         "role": "user",
         "content": [
-            {"type": "text", "text": "read the file", "cache_control": _stale_marker()}
-        ],
-    }
-    messages[2] = {
-        "role": "tool",
-        "tool_call_id": "call_1",
-        "content": [
-            {"type": "text", "text": "file body", "cache_control": _stale_marker()}
+            {
+                "type": "text",
+                "text": "read the file",
+                "cache_control": {"type": "ephemeral"},
+            }
         ],
-        "cache_control": _stale_marker(),
     }
 
-    _, out, _ = AnthropicProvider._apply_cache_control("sys", messages, None)
-
-    # Exactly one message-level breakpoint, on the LATEST non-system message.
-    marked = [
-        i
-        for i, m in enumerate(out)
-        if "cache_control" in m
-        or (
-            isinstance(m.get("content"), list)
-            and any(isinstance(b, dict) and "cache_control" in b for b in m["content"])
-        )
-    ]
-    assert marked == [3]  # final user message, string content -> wrapped block
-    # The stale message-level key on the tool result is gone.
-    assert "cache_control" not in out[2]
-    # String content at the frontier is block-wrapped, not mutated in place.
-    assert out[3]["content"] == [
-        {
-            "type": "text",
-            "text": "thanks",
-            "cache_control": {"type": "ephemeral"},
-        }
-    ]
-    assert messages[3]["content"] == "thanks"  # caller list untouched
-
-
-def test_trailing_system_note_does_not_steal_the_frontier() -> None:
-    messages = _conversation() + [{"role": "system", "content": "[note]"}]
-    _, out, _ = AnthropicProvider._apply_cache_control("sys", messages, None)
-    # Frontier marker sits on the last NON-system message, not the note.
-    # String content is block-wrapped, so check both message-level keys and
-    # content-block markers.
-    marked = [
-        i
-        for i, m in enumerate(out)
-        if "cache_control" in m
-        or (
-            isinstance(m.get("content"), list)
-            and any(isinstance(b, dict) and "cache_control" in b for b in m["content"])
-        )
-    ]
-    assert marked == [3]
+    kwargs = _build(messages)
+    out = kwargs["messages"]
 
-
-def test_all_system_history_places_no_message_marker() -> None:
-    messages = [{"role": "system", "content": "only notes"}]
-    _, out, _ = AnthropicProvider._apply_cache_control("sys", messages, None)
-    assert _markers(out) == 0
+    # Exactly one message-level breakpoint, on the LATEST message.
+    assert _marked_indices(out) == [len(out) - 1]
+    # The stale marker on the early user message is gone after conversion.
+    first = out[0]
+    assert isinstance(first.get("content"), list)
+    assert all("cache_control" not in b for b in first["content"])
 
 
 # ---------------------------------------------------------------------------
@@ -164,56 +172,43 @@ def test_all_system_history_places_no_message_marker() -> None:
 
 
 def test_marker_budget_never_exceeds_four() -> None:
-    tools = [
-        {"name": "read", "description": "", "input_schema": {}},
-        {"name": "bash", "description": "", "input_schema": {}},
-        {"name": "edit", "description": "", "input_schema": {}},
-        {"name": "mcp__github__createIssue", "description": "", "input_schema": {}},
-        {"name": "mcp__github__listIssues", "description": "", "input_schema": {}},
-    ]
-    system, out, out_tools = AnthropicProvider._apply_cache_control(
-        "base instructions", _conversation(), tools
+    kwargs = _build(
+        _conversation(),
+        _tools(
+            "read",
+            "bash",
+            "edit",
+            "mcp__github__createIssue",
+            "mcp__github__listIssues",
+        ),
+    )
+    total = _count_markers(
+        kwargs.get("system") if isinstance(kwargs.get("system"), list) else None,
+        kwargs["messages"],
+        kwargs.get("tools"),
     )
-    total = _markers(system if isinstance(system, list) else None, out, out_tools)
     assert total <= 4
     assert total >= 3  # system 1 + message 1 + at least one tool
 
 
 def test_tool_markers_cover_last_builtin_and_tail() -> None:
-    tools = [
-        {"name": "read", "description": "", "input_schema": {}},
-        {"name": "bash", "description": "", "input_schema": {}},
-        {"name": "mcp__ghidra__decompile", "description": "", "input_schema": {}},
-    ]
-    _, _, out_tools = AnthropicProvider._apply_cache_control(
-        "sys", _conversation(), tools
-    )
-    marked = [i for i, t in enumerate(out_tools or []) if "cache_control" in t]
+    kwargs = _build(_conversation(), _tools("read", "bash", "mcp__ghidra__decompile"))
+    out_tools = kwargs.get("tools") or []
+    marked = [i for i, t in enumerate(out_tools) if "cache_control" in t]
     # Last builtin (bash, index 1) and the MCP tail (index 2), never more.
     assert marked == [1, 2]
 
 
-def test_empty_or_none_tools_pass_through() -> None:
-    _, _, no_tools = AnthropicProvider._apply_cache_control(
-        "sys", _conversation(), None
-    )
-    _, _, empty_tools = AnthropicProvider._apply_cache_control(
-        "sys", _conversation(), []
-    )
-    assert no_tools is None
-    assert empty_tools == []
-
-
 # ---------------------------------------------------------------------------
-# System breakpoint + cacheHint sections
+# System + non-caching paths
 # ---------------------------------------------------------------------------
 
 
 def test_string_system_becomes_one_marked_block() -> None:
-    system, _, _ = AnthropicProvider._apply_cache_control(
-        "base instructions", _conversation(), None
+    kwargs = _build(
+        [{"role": "system", "content": "base instructions"}] + _conversation()
     )
-    assert system == [
+    assert kwargs["system"] == [
         {
             "type": "text",
             "text": "base instructions",
@@ -222,71 +217,18 @@ def test_string_system_becomes_one_marked_block() -> None:
     ]
 
 
-def test_cache_hint_stable_boundary_wins_and_hints_are_stripped() -> None:
-    system_blocks: list[dict[str, Any]] = [
-        {"type": "text", "text": "persona", "cacheHint": "stable"},
-        {"type": "text", "text": "collab preamble", "cacheHint": "stable"},
-        {"type": "text", "text": "project docs", "cacheHint": "dynamic"},
-        {"type": "text", "text": "memory index", "cacheHint": "dynamic"},
-    ]
-    system, _, _ = AnthropicProvider._apply_cache_control(
-        system_blocks, _conversation(), None
+def test_no_markers_when_caching_disabled() -> None:
+    kwargs = _build(
+        [{"role": "system", "content": "base"}] + _conversation(),
+        _tools("read", "bash", "edit"),
+        supports_caching=False,
     )
-    assert isinstance(system, list)
-    # Breakpoint on the LAST STABLE block (index 1), not the list tail.
-    marked = [i for i, b in enumerate(system) if "cache_control" in b]
-    assert marked == [1]
-    # The private hint key never survives to the API payload.
-    assert all("cacheHint" not in b for b in system)
-    # Block order and text are preserved.
-    assert [b["text"] for b in system] == [
-        "persona",
-        "collab preamble",
-        "project docs",
-        "memory index",
-    ]
+    system = kwargs.get("system")
+    assert system == "base"  # plain string, never block-wrapped on this path
+    assert _count_markers(None, kwargs["messages"], kwargs.get("tools")) == 0
+    assert "cache_control" not in system
 
 
-def test_unhinted_block_list_keeps_last_block_behavior() -> None:
-    system_blocks: list[dict[str, Any]] = [
-        {"type": "text", "text": "a"},
-        {"type": "text", "text": "b"},
-    ]
-    system, _, _ = AnthropicProvider._apply_cache_control(
-        system_blocks, _conversation(), None
-    )
-    assert isinstance(system, list)
-    assert [i for i, b in enumerate(system) if "cache_control" in b] == [1]
-
-
-def test_stale_system_markers_are_stripped_before_finalize() -> None:
-    system_blocks: list[dict[str, Any]] = [
-        {"type": "text", "text": "a", "cache_control": _stale_marker()},
-        {"type": "text", "text": "b", "cache_control": _stale_marker()},
-    ]
-    system, _, _ = AnthropicProvider._apply_cache_control(
-        system_blocks, _conversation(), None
-    )
-    assert isinstance(system, list)
-    assert [i for i, b in enumerate(system) if "cache_control" in b] == [1]
-
-
-def test_empty_system_passes_through_unchanged() -> None:
-    empty_list, _, _ = AnthropicProvider._apply_cache_control([], _conversation(), None)
-    empty_str, _, _ = AnthropicProvider._apply_cache_control("", _conversation(), None)
-    assert empty_list == []
-    assert empty_str == ""
-
-
-def test_strip_system_cache_hints_is_copy_on_write() -> None:
-    blocks = [
-        {"type": "text", "text": "a", "cacheHint": "stable"},
-        {"type": "text", "text": "b", "cacheHint": "dynamic"},
-    ]
-    stripped = AnthropicProvider._strip_system_cache_hints(blocks)
-    assert all("cacheHint" not in b for b in stripped)
-    # Input untouched (copy-on-write): the caller's history keeps its hint.
-    assert all("cacheHint" in b for b in blocks)
-    # Idempotent + str passthrough.
-    assert AnthropicProvider._strip_system_cache_hints(stripped) == stripped
-    assert AnthropicProvider._strip_system_cache_hints("plain") == "plain"
+def test_tools_absent_when_none_or_empty() -> None:
+    assert "tools" not in _build(_conversation(), None)
+    assert "tools" not in _build(_conversation(), [])
diff --git a/tests/test_cache_hint_system_blocks.py b/tests/test_cache_hint_system_blocks.py
deleted file mode 100644
index 16ed126be..000000000
--- a/tests/test_cache_hint_system_blocks.py
+++ /dev/null
@@ -1,80 +0,0 @@
-"""cacheHint hygiene on the non-Anthropic paths.
-
-``cacheHint`` is a private agent-side key (see ``core.agent_setup``) that
-marks the stable/dynamic boundary of a block-list system prompt for the
-Anthropic cache layer. Providers that do not understand it must strip it
-before the payload leaves the process, and must not lose the text it
-annotates.
-
-Contract under test:
-
-- OpenAI-compat message sanitization removes ``cacheHint`` from content
-  BLOCKS (top-level key filtering alone would leak it);
-- the Responses-API converter joins block-list system prompts into plain
-  text, ignoring ``cacheHint`` entirely — the hint key is Anthropic-only
-  plumbing and must not appear anywhere in the converted payload.
-"""
-
-from __future__ import annotations
-
-import sys
-from pathlib import Path
-from typing import Any
-
-ROOT = Path(__file__).resolve().parents[1]
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
-
-from core.providers.openai_compat import OpenAICompatProvider
-from core.providers.openai_responses.converters import convert_messages
-
-
-def _hinted_system() -> list[dict[str, Any]]:
-    return [
-        {"type": "text", "text": "persona", "cacheHint": "stable"},
-        {"type": "text", "text": "project docs", "cacheHint": "dynamic"},
-    ]
-
-
-def test_openai_compat_sanitizer_strips_cache_hint_from_blocks() -> None:
-    provider = OpenAICompatProvider(api_key="test-key")
-    messages: list[dict[str, Any]] = [
-        {"role": "system", "content": _hinted_system()},
-        {"role": "user", "content": "hello"},
-    ]
-    sanitized = provider._sanitize_messages(messages)
-
-    system_content = sanitized[0]["content"]
-    assert isinstance(system_content, list)
-    assert all("cacheHint" not in block for block in system_content)
-    # The annotated text survives — only the private key is removed.
-    assert [block["text"] for block in system_content] == ["persona", "project docs"]
-    # The original caller list is untouched (sanitization is copy-on-write).
-    assert all("cacheHint" in block for block in messages[0]["content"])
-
-
-def test_responses_converter_joins_hinted_blocks_and_drops_the_key() -> None:
-    messages: list[dict[str, Any]] = [
-        {"role": "system", "content": _hinted_system()},
-        {"role": "user", "content": "hello"},
-    ]
-    system_prompt, input_items = convert_messages(messages)
-
-    # Hint keys are Anthropic-only plumbing: the joined text is clean.
-    assert system_prompt == "persona\n\nproject docs"
-    assert "cacheHint" not in system_prompt
-    assert "cacheHint" not in str(input_items)
-    # The converter wraps string content into Responses-API input_text blocks.
-    assert input_items == [
-        {"role": "user", "content": [{"type": "input_text", "text": "hello"}]}
-    ]
-
-
-def test_responses_converter_string_system_passthrough() -> None:
-    system_prompt, input_items = convert_messages(
-        [{"role": "system", "content": "plain"}, {"role": "user", "content": "hi"}]
-    )
-    assert system_prompt == "plain"
-    assert input_items == [
-        {"role": "user", "content": [{"type": "input_text", "text": "hi"}]}
-    ]