headroom/tests/test_issue_2671_block_growth_cache.py

"""Comprehensive regression for #2671's block-growing Anthropic histories.

The provider writes cache entries only at explicit breakpoints and searches at
most 20 block boundaries backwards on the next request.  Consequently:

* a pure append must advance the breakpoint to the newest block;
* a rewritten tail must anchor at the last byte-stable leading block;
* both shapes must retain one conversation lineage across turns;
* different tools/thinking profiles must never share that lineage, because
  Anthropic renders those segments before messages in its cache key.

The small cache oracle below models those write/lookback rules.  It catches a
green-but-inert implementation: merely moving a marker in a unit-built message
is insufficient unless the real resolve -> normalize -> record sequence carries
the previous turn's state forward.
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import Any

from headroom.cache.prefix_tracker import (
    RELATION_BLOCK_APPEND,
    RELATION_BLOCK_REWRITE_TAIL,
    RELATION_DIVERGED,
    PrefixFreezeConfig,
    SessionTrackerStore,
    _strip_cache_control,
    classify_history_relation,
    extract_cache_stable_delta,
    normalize_message_cache_control,
    overlay_cached_prefix,
    segment_fingerprint,
)


def _text(text: str, *, cache: bool = False) -> dict[str, Any]:
    block: dict[str, Any] = {"type": "text", "text": text}
    if cache:
        block["cache_control"] = {"type": "ephemeral"}
    return block


def _message(blocks: list[dict[str, Any]]) -> list[dict[str, Any]]:
    return [{"role": "user", "content": blocks}]


def _pure_append(total: int) -> list[dict[str, Any]]:
    return _message([_text(f"block-{index}") for index in range(total)])


def _rewritten_tail(
    turn: int,
    churn_blocks: int,
    *,
    stable_blocks: int = 30,
    instruction: str = "instruction: summarize",
) -> list[dict[str, Any]]:
    blocks = [_text(f"stable-{index}") for index in range(stable_blocks)]
    blocks += [_text(f"turn-{turn}-changing-{index}") for index in range(churn_blocks)]
    # The captured production shape keeps a two-block identity suffix pinned at
    # the end while the blocks immediately before it are rewritten.
    blocks += [_text(instruction), _text("fixed end-of-transcript reminder")]
    return _message(blocks)


def _breakpoint(messages: list[dict[str, Any]]) -> tuple[int, int]:
    found = [
        (message_index, block_index)
        for message_index, message in enumerate(messages)
        if isinstance(message.get("content"), list)
        for block_index, block in enumerate(message["content"])
        if isinstance(block, dict) and "cache_control" in block
    ]
    assert len(found) == 1
    return found[0]


@dataclass
class _AnthropicBreakpointCache:
    """Deterministic model of Anthropic's explicit-breakpoint cache lookup."""

    entries: dict[str, int] = field(default_factory=dict)
    lookback_blocks: int = 20

    @staticmethod
    def _blocks(messages: list[dict[str, Any]]) -> list[Any]:
        blocks: list[Any] = []
        for message in messages:
            content = message.get("content")
            if isinstance(content, list):
                blocks.extend(_strip_cache_control(content))
        return blocks

    @staticmethod
    def _key(blocks: list[Any], end: int) -> str:
        return json.dumps(blocks[: end + 1], sort_keys=True, separators=(",", ":"))

    def request(self, messages: list[dict[str, Any]]) -> tuple[int, int]:
        """Return simulated ``(cache_read_blocks, cache_write_blocks)``."""
        _, breakpoint = _breakpoint(messages)
        blocks = self._blocks(messages)
        read = 0
        first = max(0, breakpoint - self.lookback_blocks + 1)
        for candidate in range(breakpoint, first - 1, -1):
            key = self._key(blocks, candidate)
            if key in self.entries:
                read = self.entries[key]
                break
        written_prefix = breakpoint + 1
        write = max(0, written_prefix - read)
        self.entries[self._key(blocks, breakpoint)] = written_prefix
        return read, write


def _record(tracker, original, forwarded, *, read=0, write=10_000):  # noqa: ANN001
    tracker.update_from_response(
        cache_read_tokens=read,
        cache_write_tokens=write,
        messages=forwarded,
        original_messages=original,
    )


def test_classifier_separates_pure_append_from_rewritten_tail() -> None:
    append = classify_history_relation(_pure_append(35), _pure_append(30))
    rewrite = classify_history_relation(_rewritten_tail(2, 5), _rewritten_tail(1, 3))

    assert append.kind == RELATION_BLOCK_APPEND
    assert append.stable_prefix_blocks == 30
    assert rewrite.kind == RELATION_BLOCK_REWRITE_TAIL
    assert rewrite.stable_prefix_blocks == 30
    assert rewrite.stable_suffix_blocks == 2


def test_rewritten_tail_requires_a_real_previous_divergence() -> None:
    """The #2702 bug classified a pure append as a rewritten tail."""
    previous = _pure_append(30)
    current = _pure_append(31)
    relation = classify_history_relation(current, previous)

    assert relation.kind == RELATION_BLOCK_APPEND
    assert relation.stable_prefix_blocks == relation.previous_block_count


def test_rewritten_tail_requires_a_two_block_identity_suffix() -> None:
    """Sibling sub-calls sharing a transcript and generic reminder must split."""
    previous = _rewritten_tail(1, 3, instruction="instruction: summarize")
    sibling = _rewritten_tail(2, 5, instruction="instruction: title")

    assert classify_history_relation(sibling, previous).kind == RELATION_DIVERGED


def test_lineage_survives_rewritten_tail_growth_and_delivers_previous_state() -> None:
    store = SessionTrackerStore(PrefixFreezeConfig(min_cached_tokens=0))
    first_tracker = None

    for turn, churn in enumerate((3, 5, 8, 11), start=1):
        original = _rewritten_tail(turn, churn)
        tracker = store.resolve_tracker("shared", "anthropic", messages=original)
        first_tracker = first_tracker or tracker
        assert tracker is first_tracker
        previous = tracker.get_last_forwarded_messages()
        if turn > 1:
            assert previous, "lineage match must deliver the previous forwarded request"
        forwarded = normalize_message_cache_control(original, previous)
        _record(tracker, original, forwarded)

    assert store.active_sessions == 1
    assert first_tracker._turn_number == 4


def test_sibling_rewritten_tail_streams_do_not_ping_pong() -> None:
    store = SessionTrackerStore()
    seen = {}
    for turn, churn in enumerate((3, 5, 8), start=1):
        for instruction in ("instruction: summarize", "instruction: title"):
            original = _rewritten_tail(turn, churn, instruction=instruction)
            tracker = store.resolve_tracker("shared", "anthropic", messages=original)
            seen.setdefault(instruction, tracker)
            assert tracker is seen[instruction]
            forwarded = normalize_message_cache_control(
                original, tracker.get_last_forwarded_messages()
            )
            _record(tracker, original, forwarded)

    assert seen["instruction: summarize"] is not seen["instruction: title"]


def test_cache_affinity_splits_identical_histories_with_different_tools() -> None:
    store = SessionTrackerStore()
    history = _pure_append(30)
    shell = segment_fingerprint({"model": "claude", "tools": [{"name": "shell"}]})
    search = segment_fingerprint({"model": "claude", "tools": [{"name": "search"}]})

    shell_tracker = store.resolve_tracker(
        "shared", "anthropic", messages=history, cache_affinity=shell
    )
    search_tracker = store.resolve_tracker(
        "shared", "anthropic", messages=history, cache_affinity=search
    )

    assert search_tracker is not shell_tracker
    assert (
        store.resolve_tracker("shared", "anthropic", messages=history, cache_affinity=shell)
        is shell_tracker
    )


def test_cache_affinity_ignores_only_cache_directive_movement() -> None:
    base = {
        "model": "claude",
        "tools": [{"name": "shell", "cache_control": {"type": "ephemeral"}}],
    }
    moved = {"model": "claude", "tools": [{"name": "shell"}]}
    changed = {"model": "claude", "tools": [{"name": "search"}]}

    assert segment_fingerprint(base) == segment_fingerprint(moved)
    assert segment_fingerprint(base) != segment_fingerprint(changed)


def test_pure_append_replays_forwarded_blocks_and_advances_breakpoint() -> None:
    previous_original = _pure_append(30)
    previous_forwarded = _message([_text(f"COMPRESSED-{index}") for index in range(30)])
    current = _pure_append(34)

    overlaid = overlay_cached_prefix(current, current, previous_original, previous_forwarded)
    normalized = normalize_message_cache_control(overlaid, previous_forwarded)

    assert [block["text"] for block in normalized[0]["content"][:30]] == [
        f"COMPRESSED-{index}" for index in range(30)
    ]
    assert [block["text"] for block in normalized[0]["content"][30:]] == [
        f"block-{index}" for index in range(30, 34)
    ]
    assert _breakpoint(normalized) == (0, 33)


def test_whole_message_delta_path_cannot_discard_appended_blocks() -> None:
    """Block appends require a splice, never an empty whole-message delta."""
    previous = _pure_append(30)

    assert extract_cache_stable_delta(_pure_append(34), previous, previous) is None


def test_cache_oracle_proves_pure_append_chains_without_rewrites() -> None:
    oracle = _AnthropicBreakpointCache()
    previous = None
    outcomes = []
    for total in (30, 34, 38, 43):
        current = _pure_append(total)
        forwarded = normalize_message_cache_control(current, previous)
        outcomes.append(oracle.request(forwarded))
        previous = forwarded

    assert outcomes == [(0, 30), (30, 4), (34, 4), (38, 5)]


def test_cache_oracle_proves_rewritten_tail_stops_perpetual_full_writes() -> None:
    oracle = _AnthropicBreakpointCache()
    previous = None
    outcomes = []
    breakpoints = []
    for turn, churn in enumerate((3, 5, 8, 11), start=1):
        current = _rewritten_tail(turn, churn)
        forwarded = normalize_message_cache_control(current, previous)
        breakpoints.append(_breakpoint(forwarded)[1])
        outcomes.append(oracle.request(forwarded))
        previous = forwarded

    # Cold turn writes its varying tail.  Turn two establishes the new stable
    # boundary; subsequent turns read it and perform no repeated full write.
    assert breakpoints == [34, 29, 29, 29]
    assert outcomes[0] == (0, 35)
    assert outcomes[1] == (0, 30)
    assert outcomes[2:] == [(30, 0), (30, 0)]


def test_relocation_kill_switch_restores_newest_block(monkeypatch) -> None:  # noqa: ANN001
    previous = normalize_message_cache_control(_rewritten_tail(1, 3))
    monkeypatch.setenv("HEADROOM_STABLE_BOUNDARY_BREAKPOINT", "0")
    current = _rewritten_tail(2, 5)

    forwarded = normalize_message_cache_control(current, previous)

    assert _breakpoint(forwarded) == (0, len(current[0]["content"]) - 1)