From 07392c462ca5e9bbe25317cf9fb29e998ee213d7 Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Mon, 31 Aug 2026 12:37:17 -0700
Subject: [PATCH 1/2] feat(cache): generalize breakpoint eligibility to
 unstable-suffix, not trailing-only
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`_count_trailing_ephemeral_messages` walked backward from the end and
stopped at the first message that was either role="tool" or NOT marked
ephemeral -- so it could only ever exclude a TRAILING run of ephemeral
messages from cache-breakpoint eligibility. The system-reminder redesign
(in-flight, amplifier-module-loop-streaming) introduces a LEADING
pre-user reminder block: `[..., block(ephemeral, ...), user(real), ...]`.
Under `ephemeral_injection_mode="tail"` + `reminder_placement="pre_user"`,
that block is regenerated per request (never persisted) but is no longer
trailing -- the old walk's count stayed 0, so a cache breakpoint could
land at or after the block, guaranteeing a cache miss on the very next
request (the exact "write every turn, read never" failure class this
provider's cache-breakpoint logic exists to prevent).

Replaces `_count_trailing_ephemeral_messages` with
`_unstable_suffix_length`, keyed on the metadata signal already present
in the data: `ephemeral=True, persisted=True` = STABLE (frozen canonical
history, written via `context.add_message`, safe to cache past);
`ephemeral=True` alone (no `persisted`) = UNSTABLE (regenerated per
request). Everything from the last unstable message through the end of
the conversation is excluded from eligibility, whether or not that
message is itself last.

This is a STRICT GENERALIZATION, verified against every shape in the
spec's own table plus a from-scratch stash-compare of the full suite:
- Tail mode today (`[..., asst, block]`): old excludes 1, new excludes 1.
  Identical.
- Tail mode + the redesign's pre-user placement (`[..., asst, block,
  user]`): old excludes 0 (bug), new excludes 2 (fixed).
- Persist mode (`[..., block(persisted), user]`): old excludes 0, new
  excludes 0. Identical (the block is genuinely frozen history).
- Persist mode, mid-loop trailing block (`[..., block(persisted)]`): old
  excludes 1 (over-conservative -- the old walk never checked
  `persisted`), new excludes 0 (improvement: a stable tail can now take a
  breakpoint).
- All-ephemeral request: old excludes len, new excludes len. Identical
  (the `eligible_upper <= 0` guard still fires).
- No metadata anywhere: unaffected either way (the loud-skip path is
  untouched).

`_find_rolling_secondary_index` (the secondary/rolling cache breakpoint)
is DELIBERATELY LEFT UNCHANGED, per the spec's own instruction -- but its
docstring is corrected here to describe the ACTUALLY VERIFIED behavior
under the new pre-user-block shape (empirically probed, not just derived
on paper): the secondary breakpoint's real benefit for this shape is a
WITHIN-A-TURN property (stable across a turn's own tool-loop iterations),
not a cross-turn one as the spec's hand-derived trace assumed -- see the
updated docstring and T-W5-04's test for the full derivation and the
concrete counter-example that led to the correction.

Tests (tests/test_prompt_cache_breakpoints.py, 28 -> 35):
- Updated: the docstring on the existing trailing-tail test now states
  explicitly that it stays a genuinely UNSTABLE shape (no `persisted`
  key), and a new persisted-metadata variant of the existing
  multi-message regression guard is added alongside it.
- New (6 tests, T-W5-01..06): unstable block before the trailing user
  excludes both from eligibility (FAILS on 833403b -- confirmed via `git
  stash`: old code stamped BOTH the block and the trailing user);
  persisted block before the user does not reduce eligibility (passes
  both before and after -- documented pin, not a fail-before proof, see
  PR description); a persisted TRAILING message may now take a
  breakpoint (FAILS on 833403b -- confirmed via `git stash`: old code
  always excluded it); the rolling secondary breakpoint's real
  within-turn-iteration behavior under the pre-user-block shape (passes
  both before and after -- `_find_rolling_secondary_index` is untouched);
  an unstable message buried behind a tool-role batch degrades safely,
  no crash; the pre-user-block shape still respects the 4-breakpoint
  hard limit.

Suite: 754 passed (747 baseline + 7 net new/changed).
`python_check`: 16 errors / 19 warnings across both touched files,
confirmed via `git stash` compare against unmodified `833403b` to be
identical, pre-existing, and unrelated to this diff (mock `.get()` typed
as list-index access, blind-except lint warnings elsewhere in the file,
etc.) -- zero new issues introduced.

Spec: reminder-redesign-spec.md, section W5. Per the spec's
dependency-ordered shipping order (section 13), W5 must land before W1
(amplifier-module-loop-streaming) -- it is a strict generalization
correct for today's shapes too, so it ships ahead safely and prevents W1
from introducing a tail-mode cache regression when it lands.

Merge note: this PR is intentionally left OPEN, not merged, pending DTU
validation per the task's DTU-gated-merge policy (see PR description).

🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier)

Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
---
 .../__init__.py                               | 139 +++++--
 tests/test_prompt_cache_breakpoints.py        | 356 +++++++++++++++++-
 2 files changed, 454 insertions(+), 41 deletions(-)

diff --git a/amplifier_module_provider_anthropic/__init__.py b/amplifier_module_provider_anthropic/__init__.py
index ca845de..2fd68a7 100644
--- a/amplifier_module_provider_anthropic/__init__.py
+++ b/amplifier_module_provider_anthropic/__init__.py
@@ -3267,8 +3267,8 @@ async def _complete_chat_request(
         # Determine ephemeral status BEFORE conversion -- Message.metadata is
         # only available on the original Message objects; _convert_messages
         # discards unrecognized keys when it rebuilds Anthropic-format dicts.
-        trailing_ephemeral_count, has_ephemeral_signal = (
-            self._count_trailing_ephemeral_messages(conversation)
+        unstable_suffix_len, has_ephemeral_signal = self._unstable_suffix_length(
+            conversation
         )
 
         # Track how many of the 4 Anthropic cache breakpoints are used, so we
@@ -3327,7 +3327,7 @@ async def _complete_chat_request(
         all_messages, conversation_breakpoints_used = (
             self._apply_conversation_cache_control(
                 all_messages,
-                trailing_ephemeral_count,
+                unstable_suffix_len,
                 has_ephemeral_signal,
                 conversation_budget,
             )
@@ -5184,46 +5184,63 @@ def _last_safe_breakpoint_index(
             idx -= 1
         return None
 
-    def _count_trailing_ephemeral_messages(
-        self, conversation: list[Message]
-    ) -> tuple[int, bool]:
-        """Count how many trailing conversation messages are ephemeral.
-
-        Ephemeral status is read from ``Message.metadata["ephemeral"]`` --
-        the same contract ``HookResult.ephemeral`` is meant to carry through
-        to the provider (see ``amplifier_core.models.HookResult``). Walking
-        from the end, the count stops at the first message that is either
-        not marked ephemeral, or has ``role == "tool"``: tool-role messages
-        can be batched together by ``_convert_messages`` (many input
-        messages -> one output message), which would break the 1:1 index
-        correspondence this count relies on. Ephemeral injections are never
-        tool-role in the current architecture (HookResult.context_injection_role
-        is one of "system"/"user"/"assistant"), so this is a hard stop, not a
-        heuristic guess.
+    def _unstable_suffix_length(self, conversation: list[Message]) -> tuple[int, bool]:
+        """How many trailing entries must be excluded from breakpoint eligibility.
+
+        A message is UNSTABLE when it is marked ephemeral but not persisted --
+        i.e. regenerated per request, so any cached prefix containing it can
+        never be reproduced. Everything from the last unstable message to the
+        end is excluded, whether or not the unstable message is itself last
+        (the pre-user reminder block introduced by the system-reminder
+        redesign is not: it sits BEFORE the real user message, not after it).
+
+        Ephemeral AND persisted is STABLE: it was written into canonical
+        history via context.add_message and is byte-frozen from then on
+        (the persist-mode reminder block, and the orchestrator's own
+        budget-warning message, both qualify).
+
+        This is a strict generalization of the prior
+        ``_count_trailing_ephemeral_messages`` (any ephemeral message,
+        regardless of position, disqualified the tail): that method's
+        behavior is reproduced exactly whenever ephemeral content only ever
+        appears at the true tail (today's shape, and the redesign's own
+        tail-injection-mode shape), and additionally handles a leading
+        ephemeral-but-unstable block correctly, which the old walk could not
+        express at all.
+
+        The role == "tool" hard stop is retained verbatim from
+        _count_trailing_ephemeral_messages: _convert_messages batches
+        consecutive tool messages many-to-one, which breaks the 1:1 index
+        correspondence a count-from-the-end relies on. Stopping early is safe
+        here because view-only (unstable) injections only ever exist for the
+        CURRENT request -- they are never in canonical history, so none can be
+        buried behind an earlier tool batch.
 
         Returns:
-            (trailing_ephemeral_count, has_any_metadata_signal). The second
-            value tells the caller whether *any* message in the conversation
-            carries a metadata dict at all -- i.e. whether "not marked
-            ephemeral" is a trustworthy negative, or simply means nothing in
-            this deployment ever populates the field.
+            (excluded, has_any_metadata_signal). ``excluded`` is a count from
+            the end (never an absolute index -- ``all_messages`` combines
+            context-prefix messages with converted conversation messages, so
+            pre-conversion indices do not map 1:1 onto it).
+            ``has_any_metadata_signal`` tells the caller whether *any*
+            message in the conversation carries a metadata dict at all --
+            i.e. whether "not unstable" is a trustworthy negative, or simply
+            means nothing in this deployment ever populates the field.
         """
         has_any_metadata_signal = any(bool(m.metadata) for m in conversation)
 
-        count = 0
-        for msg in reversed(conversation):
+        excluded = 0
+        for walked, msg in enumerate(reversed(conversation), start=1):
             if msg.role == "tool":
                 break
-            if bool((msg.metadata or {}).get("ephemeral")):
-                count += 1
-                continue
-            break
-        return count, has_any_metadata_signal
+            md = msg.metadata or {}
+            if md.get("ephemeral") and not md.get("persisted"):
+                excluded = walked
+        return excluded, has_any_metadata_signal
 
     def _apply_conversation_cache_control(
         self,
         all_messages: list[dict[str, Any]],
-        trailing_ephemeral_count: int,
+        unstable_suffix_len: int,
         has_ephemeral_signal: bool,
         remaining_budget: int,
     ) -> tuple[list[dict[str, Any]], int]:
@@ -5243,8 +5260,12 @@ def _apply_conversation_cache_control(
 
         The fix has two parts:
 
-        1. Never place a breakpoint on a message known to be ephemeral --
-           walk backward past ``trailing_ephemeral_count`` messages first.
+        1. Never place a breakpoint on a message known to be unstable
+           (ephemeral and not persisted -- regenerated per request) --
+           walk backward past ``unstable_suffix_len`` messages first. Unlike
+           the prior trailing-only walk, this correctly excludes an unstable
+           message wherever it sits in the eligible window, not only when
+           it is literally last (see ``_unstable_suffix_length``).
         2. Use the *two* breakpoints Anthropic's docs describe for
            multi-turn conversations: one at the current stable boundary
            ("primary") and one right before the previous real user turn
@@ -5257,13 +5278,13 @@ def _apply_conversation_cache_control(
             all_messages: Anthropic-formatted message array (context-prefix
                 messages followed by conversation messages), mutated in
                 place.
-            trailing_ephemeral_count: number of trailing entries of the
-                conversation region known to be ephemeral (see
-                ``_count_trailing_ephemeral_messages``).
+            unstable_suffix_len: number of trailing entries of the
+                conversation region excluded from breakpoint eligibility
+                (see ``_unstable_suffix_length``).
             has_ephemeral_signal: whether ephemeral status could be
                 determined at all for this request (see above). If False,
-                "not marked ephemeral" cannot be trusted as a real signal,
-                and placing breakpoints on unverified content would risk
+                "not unstable" cannot be trusted as a real signal, and
+                placing breakpoints on unverified content would risk
                 repeating the exact bug this method exists to fix.
             remaining_budget: cache breakpoints left before hitting
                 Anthropic's 4-breakpoint hard limit (system + tools may
@@ -5318,7 +5339,7 @@ def _apply_conversation_cache_control(
             )
             return all_messages, 0
 
-        eligible_upper = len(all_messages) - trailing_ephemeral_count
+        eligible_upper = len(all_messages) - unstable_suffix_len
         if eligible_upper <= 0:
             logger.warning(
                 "[PROVIDER] Prompt caching: every message in this request is "
@@ -5372,6 +5393,44 @@ def _find_rolling_secondary_index(
         breakpoints keeps landing on a previously-cached boundary as the
         conversation grows, per Anthropic's documented multi-turn caching
         pattern.
+
+        Deliberately left alone under the system-reminder redesign's
+        pre-user reminder block (a stable, persisted, `role="user"` message
+        written immediately before the turn's real user message -- see
+        ``_unstable_suffix_length``). Verified empirically (see
+        tests/test_prompt_cache_breakpoints.py's T-W5-04 test), not just
+        derived on paper -- the actual mechanism is subtler than "the block
+        becomes the previous turn's primary":
+
+        At iteration 1 of a fresh turn N+1 (request ends `[..., block N+1,
+        user N+1]`, no assistant reply yet), `primary_idx` IS `user N+1`'s
+        own index. This method's search starts AT `primary_idx` itself, and
+        since that message is `role="user"` and not a tool_result batch, it
+        matches on the very first loop iteration -- `last_user_turn_idx`
+        resolves to `user N+1`'s own index (not the block's), and the
+        secondary is `safe(last_user_turn_idx - 1)`, which lands ON `block
+        N+1` (it sits directly before `user N+1`).
+
+        Once the model replies (iteration 2+: `[..., block N+1, user N+1,
+        assistant N+1(tool_use), tool_result, ...]`), `primary_idx` advances
+        into the tool_result batch. The search now walks PAST the
+        tool_result (excluded: it IS a tool_result batch) and past the
+        assistant message, and matches `user N+1` the same way -- so the
+        secondary lands on `block N+1` AGAIN, at the same position, across
+        every iteration of the SAME turn. That is the real rolling-overlap
+        property this reminder-block shape gets from this method, unchanged:
+        a consistent secondary target across a turn's own tool-loop
+        iterations. (A cross-TURN overlap -- turn N+1's secondary matching
+        turn N's OWN primary from the request that generated it -- does NOT
+        hold for this shape: that hypothetical prior request also ends in
+        `[..., block N, user N]` with no assistant reply yet, so its own
+        primary is `user N`'s index, not `block N`'s -- there is no shared
+        position between the two calls in that comparison. This is a
+        pre-existing property of "primary lands on a user-role message
+        when nothing follows it yet", not something the system-reminder
+        redesign changes.) Do not "fix" this to skip reminder blocks --
+        skipping them here would break the WITHIN-turn overlap that
+        genuinely exists.
         """
         last_user_turn_idx: int | None = None
         for idx in range(min(primary_idx, eligible_upper - 1), -1, -1):
diff --git a/tests/test_prompt_cache_breakpoints.py b/tests/test_prompt_cache_breakpoints.py
index 4d7f7e8..95c8d49 100644
--- a/tests/test_prompt_cache_breakpoints.py
+++ b/tests/test_prompt_cache_breakpoints.py
@@ -34,7 +34,6 @@
     ToolCallBlock,
     ToolSpec,
 )
-
 from amplifier_module_provider_anthropic import AnthropicProvider
 from tests._helpers import DummyResponse
 
@@ -122,10 +121,32 @@ def _ephemeral_tail(text: str) -> Message:
     role from `context_injection_role` (default/observed: "user"), content
     is the injected text, and -- once the necessary upstream plumbing fix
     is applied -- `metadata={"ephemeral": True}`.
+
+    This is UNSTABLE under `_unstable_suffix_length` (ephemeral, no
+    `persisted` key) -- regenerated per request, e.g. loop-streaming's
+    `ephemeral_injection_mode="tail"` path, or the view-only splice for
+    `reminder_placement="pre_user"` in tail mode.
     """
     return Message(role="user", content=text, metadata={"ephemeral": True})
 
 
+def _persisted_reminder(text: str) -> Message:
+    """Build a message shaped like the system-reminder redesign's
+    persist-mode reminder block (loop-streaming's `ephemeral_injection_mode
+    ="persist"` path): written into canonical history via
+    `context.add_message(...)`, so it is byte-frozen from then on.
+
+    This is STABLE under `_unstable_suffix_length`
+    (`ephemeral=True, persisted=True`) -- safe to cache past, unlike
+    `_ephemeral_tail`'s unstable shape.
+    """
+    return Message(
+        role="user",
+        content=text,
+        metadata={"ephemeral": True, "persisted": True},
+    )
+
+
 def _run(provider: AnthropicProvider, request: ChatRequest) -> dict:
     params = _capture_params(provider)
 
@@ -198,6 +219,11 @@ def test_never_exceeds_four_breakpoints_even_with_many_turns():
 
 
 def test_breakpoint_never_lands_on_ephemeral_tail_message():
+    """Still valid under `_unstable_suffix_length` (system-reminder redesign,
+    W5): `_ephemeral_tail` carries no `persisted` key, so this remains an
+    UNSTABLE trailing message -- the mid-loop `ephemeral_injection_mode=
+    "tail"` shape (or the `reminder_placement="pre_user"` tail-mode splice,
+    which is view-only and therefore never persisted either)."""
     provider = _make_provider()
 
     messages: list[Message] = [Message(role="system", content="System prompt.")]
@@ -598,6 +624,56 @@ def test_multi_message_with_ephemeral_metadata_places_breakpoints_as_before(capl
     )
 
 
+def test_multi_message_with_persisted_ephemeral_metadata_places_breakpoints_as_before(
+    caplog,
+):
+    """Same regression guard as above, with a `persisted: True` variant
+    (system-reminder redesign, W5/T-W5-02): a persist-mode reminder block
+    sitting immediately before the trailing user message must not reduce
+    breakpoint eligibility at all -- it is genuinely frozen canonical
+    history, not regenerated content. Numerically this shape produces the
+    SAME `eligible_upper` as the pre-existing `_ephemeral_tail` (unstable)
+    variant above under BOTH the old and new algorithms here (the old
+    trailing-only walk never even reaches this position, since the very
+    last message is the real, non-ephemeral user text) -- pinning that this
+    stays true is the point: a future, less careful generalization could
+    easily start treating any `ephemeral=True` message as exclusion-worthy
+    regardless of `persisted`, which would silently regress this exact
+    case."""
+    provider = _make_provider()
+
+    messages: list[Message] = [Message(role="system", content="System prompt.")]
+    for i in range(3):
+        messages.extend(_turn(f"question {i}", f"answer {i}"))
+    messages.append(_persisted_reminder("<system-reminders>...</system-reminders>"))
+    messages.append(Message(role="user", content="the real ask"))
+
+    request = ChatRequest(messages=messages)
+
+    import logging
+
+    with caplog.at_level(logging.WARNING, logger="amplifier_module_provider_anthropic"):
+        params = _run(provider, request)
+
+    own_records = [
+        r for r in caplog.records if r.name == "amplifier_module_provider_anthropic"
+    ]
+    assert not any(r.levelno >= logging.WARNING for r in own_records)
+
+    found_breakpoint = False
+    for msg in params["messages"]:
+        content = msg.get("content")
+        if isinstance(content, list) and any(
+            isinstance(b, dict) and "cache_control" in b for b in content
+        ):
+            found_breakpoint = True
+    assert found_breakpoint, (
+        "expected at least one conversation-region cache breakpoint even "
+        "with a persisted reminder block immediately before the trailing "
+        "user message"
+    )
+
+
 # ---------------------------------------------------------------------------
 # Extended (1h) TTL opt-in for the stable system/tools region
 # ---------------------------------------------------------------------------
@@ -1221,3 +1297,281 @@ def test_breakpoint_never_lands_on_a_redacted_thinking_block():
     assert not offenders, (
         f"cache_control landed on a redacted_thinking block: {offenders}"
     )
+
+
+# ---------------------------------------------------------------------------
+# System-reminder redesign, W5: `_unstable_suffix_length` generalization.
+#
+# Prior to this patch, `_count_trailing_ephemeral_messages` walked backward
+# from the end and stopped at the first message that was either role="tool"
+# or NOT marked ephemeral -- i.e. it could only ever exclude a TRAILING run
+# of ephemeral messages. The system-reminder redesign introduces a leading
+# (pre-user) reminder block: `[..., block(ephemeral, ...), user(real), ...]`.
+# `_unstable_suffix_length` distinguishes STABLE (ephemeral AND persisted --
+# frozen canonical history) from UNSTABLE (ephemeral, NOT persisted --
+# regenerated per request) and excludes everything from the last unstable
+# message through the end, whether or not that message is itself last.
+# ---------------------------------------------------------------------------
+
+
+def test_unstable_block_before_trailing_user_excludes_both_from_eligibility():
+    """T-W5-01 (the core W5 bug). An unstable (ephemeral, not persisted)
+    block sitting BEFORE the trailing real user message -- the shape
+    produced by loop-streaming's `ephemeral_injection_mode="tail"` +
+    `reminder_placement="pre_user"` -- must exclude BOTH the block and the
+    user message that follows it from breakpoint eligibility. A breakpoint
+    landing AT OR AFTER the block would still cache the block's
+    regenerated-per-request bytes as part of the same prefix, guaranteeing
+    a miss on the very next request.
+
+    FAILS on the prior trailing-only walk: the trailing message here is the
+    REAL user text (not ephemeral), so the old walk's count stayed 0 --
+    verified via `git stash` against unmodified `833403b`: the old code
+    placed its PRIMARY breakpoint directly on the trailing user message
+    (the last "safe" position at the old, unreduced `eligible_upper`), and
+    its SECONDARY breakpoint landed directly on the unstable block itself
+    (`_find_rolling_secondary_index` treats the primary's own user-role
+    message as "the most recent real user turn" and steps back exactly one
+    message -- straight onto the block). Both assertions below fail on
+    `833403b`; both pass after this patch."""
+    provider = _make_provider()
+
+    messages: list[Message] = [Message(role="system", content="System prompt.")]
+    messages.extend(_turn("question 0", "answer 0"))
+    messages.append(_ephemeral_tail("<system-reminders>...</system-reminders>"))
+    messages.append(Message(role="user", content="the real ask"))
+
+    request = ChatRequest(messages=messages)
+    params = _run(provider, request)
+
+    sent_messages = params["messages"]
+    block_msg = sent_messages[-2]
+    user_msg = sent_messages[-1]
+
+    def _is_stamped(msg: dict) -> bool:
+        content = msg.get("content")
+        return isinstance(content, list) and any(
+            isinstance(b, dict) and "cache_control" in b for b in content
+        )
+
+    assert not _is_stamped(block_msg), (
+        "breakpoint must not land on the unstable (regenerated-per-request) "
+        "reminder block"
+    )
+    assert not _is_stamped(user_msg), (
+        "breakpoint must not land on the real user message either -- it "
+        "directly follows the unstable block, so a breakpoint here would "
+        "still cache the block's regenerated bytes as part of the same "
+        "prefix"
+    )
+    assert any(_is_stamped(m) for m in sent_messages[:-2]), (
+        "expected a breakpoint on earlier, genuinely stable content instead"
+    )
+
+
+def test_persisted_block_before_user_does_not_reduce_eligibility():
+    """T-W5-02. An ephemeral AND persisted block immediately before the
+    trailing user message (loop-streaming's `ephemeral_injection_mode=
+    "persist"` + `reminder_placement="pre_user"` turn-start shape) must NOT
+    reduce breakpoint eligibility -- it is frozen canonical history, safe
+    to cache past. Numerically this shape is invisible to the OLD
+    trailing-only walk too (the last message is the real user text, not
+    ephemeral, so the old walk's count was already 0 here) -- confirmed via
+    `git stash` against `833403b`: this test passes on both. Recorded here
+    as a deliberate documentation pin (not a fail-before proof; see the PR
+    description for this spec-ambiguity resolution), so a future, less
+    careful generalization that starts excluding ANY `ephemeral=True`
+    message regardless of `persisted` cannot silently regress this exact,
+    extremely common shape (every persist-mode turn looks like this)."""
+    provider = _make_provider()
+
+    messages: list[Message] = [Message(role="system", content="System prompt.")]
+    messages.extend(_turn("question 0", "answer 0"))
+    messages.append(_persisted_reminder("<system-reminders>...</system-reminders>"))
+    messages.append(Message(role="user", content="the real ask"))
+
+    request = ChatRequest(messages=messages)
+    params = _run(provider, request)
+
+    sent_messages = params["messages"]
+
+    def _is_stamped(msg: dict) -> bool:
+        content = msg.get("content")
+        return isinstance(content, list) and any(
+            isinstance(b, dict) and "cache_control" in b for b in content
+        )
+
+    assert any(_is_stamped(m) for m in sent_messages), (
+        "expected at least one conversation-region cache breakpoint"
+    )
+
+
+def test_persisted_trailing_message_may_take_a_breakpoint():
+    """T-W5-03. A persisted-ephemeral message that IS the trailing
+    (mid-loop) message -- e.g. loop-streaming's persist-mode reminder block
+    at turn start, before the next iteration appends anything further --
+    may now take a breakpoint. This removes the prior over-conservatism:
+    the OLD walk excluded ANY trailing ephemeral message regardless of
+    `persisted`, so a genuinely stable persisted-ephemeral tail could never
+    be cached. FAILS on `833403b` (confirmed via `git stash`): the old code
+    always excluded this trailing message from eligibility, so it was never
+    stamped; after this patch it is eligible and gets stamped like any
+    other stable trailing content."""
+    provider = _make_provider()
+
+    messages: list[Message] = [Message(role="system", content="System prompt.")]
+    for i in range(3):
+        messages.extend(_turn(f"question {i}", f"answer {i}"))
+    messages.append(_persisted_reminder("<system-reminders>...</system-reminders>"))
+
+    request = ChatRequest(messages=messages)
+    params = _run(provider, request)
+
+    sent_messages = params["messages"]
+    last_msg = sent_messages[-1]
+    content = last_msg.get("content")
+    assert isinstance(content, list) and any(
+        isinstance(b, dict) and "cache_control" in b for b in content
+    ), (
+        "expected the persisted-ephemeral trailing message to be eligible "
+        "for a cache breakpoint (the over-conservatism removed by W5)"
+    )
+
+
+def test_rolling_secondary_stays_on_pre_user_block_across_a_turns_own_iterations():
+    """T-W5-04 (pins §W5.3 -- `_find_rolling_secondary_index` is
+    deliberately left UNCHANGED). Verified empirically (see the docstring
+    on `_find_rolling_secondary_index` itself for the full derivation, and
+    the PR description for the spec-ambiguity resolution): the rolling
+    secondary breakpoint's real, verified benefit for a pre-user persisted
+    reminder block is a WITHIN-TURN property, not a cross-turn one.
+
+    Iteration 1 of a fresh turn (request ends `[..., block, user]`, no
+    assistant reply yet): `primary_idx` is `user`'s own index, and
+    `_find_rolling_secondary_index`'s search matches on that very message
+    at the first loop step, so the secondary is `safe(primary_idx - 1)` --
+    landing ON the persisted block that sits directly before it.
+
+    Iteration 2+ of the SAME turn (after a tool call: `[..., block, user,
+    assistant(tool_use), tool_result]`): `primary_idx` advances into the
+    tool_result batch; the search walks past it (excluded: it IS a
+    tool_result batch) and past the assistant message, matches `user` the
+    same way, and again returns `safe(user_idx - 1)` -- the SAME block,
+    same position. The secondary breakpoint is therefore stable across a
+    turn's own tool-loop iterations, giving a genuine rolling cache hit on
+    the reminder block.
+
+    Not a fail-before proof (see PR description for why): `_find_rolling_
+    secondary_index` is untouched by this patch, and this exact behavior is
+    unaffected by `_unstable_suffix_length` either way (the block here is
+    persisted, so neither the old nor the new eligibility walk treats it as
+    unstable) -- confirmed via `git stash` against `833403b`. Recorded as a
+    deliberate pin per §W5.3's explicit instruction not to "fix" this
+    method."""
+    shared_prefix = [
+        Message(role="system", content="System prompt."),
+        _persisted_reminder("<system-reminders>t1</system-reminders>"),
+        Message(role="user", content="question 0"),
+        Message(role="assistant", content="answer 0"),
+        _persisted_reminder("<system-reminders>t2</system-reminders>"),
+        Message(role="user", content="question 1"),
+    ]
+
+    def _cached_texts(params: dict) -> set[str]:
+        out = set()
+        for msg in params["messages"]:
+            content = msg.get("content")
+            if isinstance(content, list):
+                for block in content:
+                    if isinstance(block, dict) and "cache_control" in block:
+                        out.add(block.get("text", ""))
+        return out
+
+    # Iteration 1: request ends in [block_t2, question_1] -- no reply yet.
+    params_iter1 = _run(_make_provider(), ChatRequest(messages=shared_prefix))
+    breakpoints_iter1 = _cached_texts(params_iter1)
+
+    # Iteration 2: the model has now called a tool for THIS SAME turn.
+    messages_iter2 = shared_prefix + [
+        Message(
+            role="assistant",
+            content=[ToolCallBlock(id="call_1", name="lookup", input={})],
+        ),
+        Message(role="tool", content="tool result", tool_call_id="call_1"),
+    ]
+    params_iter2 = _run(_make_provider(), ChatRequest(messages=messages_iter2))
+    breakpoints_iter2 = _cached_texts(params_iter2)
+
+    assert breakpoints_iter1, "iteration 1 should have placed at least one breakpoint"
+    assert breakpoints_iter2, "iteration 2 should have placed at least one breakpoint"
+    overlap = breakpoints_iter1 & breakpoints_iter2
+    assert overlap, (
+        "the two iterations of the SAME turn should share at least one "
+        "cache breakpoint (the pre-user reminder block) -- otherwise the "
+        "cache can never be hit within a single tool-loop turn"
+    )
+    assert "<system-reminders>t2</system-reminders>" in overlap, (
+        f"expected the shared breakpoint to be the turn's own pre-user "
+        f"reminder block specifically; got overlap={overlap}"
+    )
+
+
+def test_unstable_message_behind_a_tool_batch_degrades_safely():
+    """T-W5-05. An unstable message positioned BEHIND (further from the
+    tail than) a trailing tool-role batch must not crash the walk, and
+    placement must degrade safely -- not raise, not infinite-loop. The
+    role=="tool" hard stop fires on the very first (trailing-most) message
+    here, before any unstable content is even examined, exactly like the
+    prior implementation's identical hard stop. This shape is not expected
+    to occur in practice (unstable injections only ever exist for the
+    CURRENT request, never behind canonical tool-call history), but the
+    method must not misbehave if it does."""
+    provider = _make_provider()
+
+    messages: list[Message] = [Message(role="system", content="System prompt.")]
+    messages.extend(_turn("question 0", "answer 0"))
+    messages.append(_ephemeral_tail("<system-reminders>buried</system-reminders>"))
+    messages.append(
+        Message(
+            role="assistant",
+            content=[ToolCallBlock(id="call_1", name="lookup", input={})],
+        )
+    )
+    messages.append(Message(role="tool", content="tool result", tool_call_id="call_1"))
+
+    request = ChatRequest(messages=messages)
+    params = _run(provider, request)  # must not raise
+
+    sent_messages = params["messages"]
+    found_breakpoint = any(
+        isinstance(msg.get("content"), list)
+        and any(isinstance(b, dict) and "cache_control" in b for b in msg["content"])
+        for msg in sent_messages
+    )
+    assert found_breakpoint, (
+        "expected placement to still succeed (degrade safely) rather than "
+        "silently place zero breakpoints"
+    )
+
+
+def test_never_exceeds_four_breakpoints_with_pre_user_block():
+    """T-W5-06 (must not regress). The new pre-user-block shape must still
+    respect Anthropic's hard 4-breakpoint limit, exactly like the existing
+    ceiling tests for the no-block shape."""
+    provider = _make_provider()
+
+    messages: list[Message] = [
+        Message(role="system", content="You are a helpful assistant.")
+    ]
+    for i in range(6):
+        messages.extend(_turn(f"question {i}", f"answer {i}"))
+    messages.append(_persisted_reminder("<system-reminders>...</system-reminders>"))
+    messages.append(Message(role="user", content="the real ask"))
+
+    request = ChatRequest(
+        messages=messages,
+        tools=[_long_tool_spec("tool_a"), _long_tool_spec("tool_b")],
+    )
+    params = _run(provider, request)
+
+    assert _count_cache_control_blocks(params) <= 4

From d5330dbb57d070cfa529bb463f0ca71ad1683920 Mon Sep 17 00:00:00 2001
From: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
Date: Mon, 31 Aug 2026 14:47:06 -0700
Subject: [PATCH 2/2] test: verify breakpoint eligibility unaffected by
 loop-streaming's D2 fix
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The 20260831-rr treatment-validation wave's D2 finding (envelope
accumulation) was fixed in amplifier-module-loop-streaming by writing
subsequent same-turn persisted reminder messages WITHOUT the repeated
descriptive boilerplate header -- a change to message CONTENT only,
never to metadata or role/position.

_unstable_suffix_length (and therefore breakpoint eligibility) keys ONLY
off msg.role and msg.metadata -- never msg.content -- so this provider
needs no code change for D2. This test pins that invariant explicitly:
a header-less persisted reminder message must be treated identically to
a full-header one (both STABLE, both breakpoint-eligible), rather than
leaving it as an inference from reading the source.

No functional change to this provider. D1 (the anthropic prompt-cache
collapse, fixed in hooks-status-context and routing-matrix) is also
independently confirmed to require no change here: the cache-breakpoint
walk operates on the MESSAGE list, entirely orthogonal to the
system-prompt-factory-wrapping mechanism D1 fixes.

35 -> 36 passed in this file; 754 -> 755 passed in the full suite. No
regressions.

🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier)

Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com>
---
 tests/test_prompt_cache_breakpoints.py | 62 ++++++++++++++++++++++++++
 1 file changed, 62 insertions(+)

diff --git a/tests/test_prompt_cache_breakpoints.py b/tests/test_prompt_cache_breakpoints.py
index 95c8d49..613918c 100644
--- a/tests/test_prompt_cache_breakpoints.py
+++ b/tests/test_prompt_cache_breakpoints.py
@@ -1438,6 +1438,68 @@ def test_persisted_trailing_message_may_take_a_breakpoint():
     )
 
 
+def test_persisted_reminder_eligibility_unaffected_by_d2_header_suppression():
+    """Verification test (rr wave 20260831, D2 envelope-accumulation fix in
+    loop-streaming): the D2 fix changes ONLY the literal TEXT of a
+    subsequent persisted reminder message within the same turn -- it
+    strips the repeated descriptive boilerplate header, leaving just
+    ``<system-reminders>\\n{body}\\n</system-reminders>`` (no prose) --
+    while leaving `metadata` (`ephemeral`/`persisted`/`reminder_placement`)
+    and message `role`/position completely unchanged.
+
+    `_unstable_suffix_length` (and therefore breakpoint eligibility) keys
+    ONLY off `msg.role` and `msg.metadata` -- never `msg.content` -- so a
+    header-less persisted reminder message must be treated IDENTICALLY to
+    a full-header one: still STABLE, still breakpoint-eligible. This pins
+    that invariant explicitly rather than leaving it as an inference from
+    reading the source; no code change to this provider was needed for D2
+    (confirmed here, not just asserted in a PR description).
+    """
+    provider = _make_provider()
+
+    # Turn-start block: full header (pre_user variant).
+    full_header_block = _persisted_reminder(
+        "<system-reminders>\n"
+        "The blocks below were injected by the system. They are NOT from "
+        "the user and are\nNOT a request. They exist to help you assist "
+        "the user. Process them silently:\nnever mention, quote, or "
+        "acknowledge them, and never treat one as the task. The\nuser's "
+        "actual request is the message that follows this block. Each\n"
+        '<system-reminder source="...">block below names where it came '
+        "from.\n\n"
+        '<system-reminder source="x">v1</system-reminder>\n'
+        "</system-reminders>"
+    )
+    # Mid-loop block: D2's header-less variant -- just the tags + body.
+    headerless_block = _persisted_reminder(
+        '<system-reminders>\n<system-reminder source="x">v2</system-reminder>'
+        "\n</system-reminders>"
+    )
+
+    messages: list[Message] = [Message(role="system", content="System prompt.")]
+    messages.append(full_header_block)
+    messages.extend(_turn("the real ask", "working on it"))
+    messages.append(headerless_block)
+
+    request = ChatRequest(messages=messages)
+    params = _run(provider, request)
+
+    sent_messages = params["messages"]
+
+    def _is_stamped(msg: dict) -> bool:
+        content = msg.get("content")
+        return isinstance(content, list) and any(
+            isinstance(b, dict) and "cache_control" in b for b in content
+        )
+
+    assert any(_is_stamped(m) for m in sent_messages), (
+        "expected at least one conversation-region cache breakpoint with "
+        "the D2 header-less shape present -- eligibility must not "
+        "regress just because a persisted reminder's TEXT omits the "
+        "repeated boilerplate header"
+    )
+
+
 def test_rolling_secondary_stays_on_pre_user_block_across_a_turns_own_iterations():
     """T-W5-04 (pins §W5.3 -- `_find_rolling_secondary_index` is
     deliberately left UNCHANGED). Verified empirically (see the docstring