Harden cache validation reporting and TTL analysis

benchmarks/cache_bust_trace_report.py

@@ -0,0 +1,356 @@
+#!/usr/bin/env python3
+"""Trace and report concrete cache-busting turns from local Claude session replays."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+
+DEFAULT_OUTPUT_DIR = Path("benchmark_results") / "cache_bust_trace"
+
+
+@dataclass
+class BustEvent:
+    branch: str
+    mode: str
+    session_id: str
+    project: str
+    request_id: str
+    timestamp: str
+    first_diff_index: int | None
+    prev_len: int
+    curr_len: int
+    prev_msg: dict[str, Any] | None
+    curr_msg: dict[str, Any] | None
+    prev_tail: list[dict[str, Any]]
+    curr_tail: list[dict[str, Any]]
+    retroactive_rewrite: bool
+
+
+def _run_git(args: list[str], cwd: Path) -> str:
+    completed = subprocess.run(
+        ["git", *args],
+        cwd=cwd,
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return completed.stdout.strip()
+
+
+def _ref_slug(ref: str) -> str:
+    return "".join(ch if ch.isalnum() else "-" for ch in ref).strip("-").lower() or "ref"
+
+
+def _first_diff_index(prev: list[dict[str, Any]], curr: list[dict[str, Any]]) -> int | None:
+    for i, (a, b) in enumerate(zip(prev, curr)):
+        if a != b:
+            return i
+    if len(prev) != len(curr):
+        return min(len(prev), len(curr))
+    return None
+
+
+def _trace_branch(
+    repo_root: Path,
+    ref: str,
+    label: str,
+    *,
+    recent_turns_per_session: int,
+    max_events_per_mode: int = 10,
+) -> list[BustEvent]:
+    worktree_root = Path(tempfile.mkdtemp(prefix="headroom-bust-trace-"))
+    worktree_dir = worktree_root / _ref_slug(label)
+    _run_git(["worktree", "add", "--detach", str(worktree_dir), ref], repo_root)
+    try:
+        env = os.environ.copy()
+        env["PYTHONPATH"] = str(worktree_dir)
+        code = """
+import copy, json
+from datetime import timedelta
+from pathlib import Path
+import importlib.util
+import os
+import sys
+
+module_path = Path(os.environ['BUST_TRACE_SCRIPT'])
+spec = importlib.util.spec_from_file_location('branch_benchmark', module_path)
+mod = importlib.util.module_from_spec(spec)
+assert spec and spec.loader
+sys.modules[spec.name] = mod
+spec.loader.exec_module(mod)
+
+PROXY_MODE_CACHE = mod.PROXY_MODE_CACHE
+PROXY_MODE_TOKEN = mod.PROXY_MODE_TOKEN
+PrefixCacheTracker = mod.PrefixCacheTracker
+_apply_mode_to_messages = mod._apply_mode_to_messages
+_cache_gap_within_ttl = mod._cache_gap_within_ttl
+_rewrite_scope = mod._rewrite_scope
+get_tokenizer = mod.get_tokenizer
+load_session_replay = mod.load_session_replay
+select_session_files = mod.select_session_files
+trim_replay_to_recent_turns = mod.trim_replay_to_recent_turns
+_make_proxy = mod._make_proxy
+from headroom.cache.compression_cache import CompressionCache
+
+ROOT = Path.home() / '.claude' / 'projects'
+TTL = timedelta(minutes=5)
+recent_turns_per_session = int(__import__('os').environ['BUST_TRACE_RECENT'])
+max_events_per_mode = int(__import__('os').environ['BUST_TRACE_MAX'])
+
+def first_diff_index(prev, curr):
+    for i, (a, b) in enumerate(zip(prev, curr)):
+        if a != b:
+            return i
+    if len(prev) != len(curr):
+        return min(len(prev), len(curr))
+    return None
+
+def trace_mode(mode):
+    proxy = _make_proxy(mode)
+    session_files = select_session_files(ROOT)
+    events = []
+    for session_file in session_files:
+        replay = load_session_replay(session_file)
+        if replay is None:
+            continue
+        replay = trim_replay_to_recent_turns(replay, recent_turns_per_session)
+        prefix_tracker = PrefixCacheTracker('anthropic')
+        comp_cache = CompressionCache() if mode == PROXY_MODE_TOKEN else None
+        conversation = []
+        conversation_token_total = 0
+        previous_forwarded = []
+        previous_original_context = None
+        previous_forwarded_context = None
+        previous_timestamp = None
+        pending = None
+        for turn in replay.turns:
+            tokenizer = get_tokenizer(turn.model)
+            turn_input_token_total = sum(tokenizer.count_message(msg) for msg in turn.input_messages)
+            prior_context_message_count = len(conversation)
+            conversation.extend(turn.input_messages)
+            raw_input_tokens = conversation_token_total + turn_input_token_total
+            forwarded = _apply_mode_to_messages(
+                proxy, mode, conversation,
+                model=turn.model, prefix_tracker=prefix_tracker, comp_cache=comp_cache,
+                previous_original_messages=previous_original_context,
+                previous_forwarded_messages=previous_forwarded_context,
+            )
+            if pending is not None:
+                eligible = _cache_gap_within_ttl(pending.turn.timestamp, previous_timestamp, ttl=TTL)
+                if eligible and previous_forwarded:
+                    prefix_preserved = (
+                        len(pending.forwarded) >= len(previous_forwarded)
+                        and pending.forwarded[: len(previous_forwarded)] == previous_forwarded
+                    )
+                    if not prefix_preserved:
+                        idx = first_diff_index(previous_forwarded, pending.forwarded)
+                        _, retro = _rewrite_scope(
+                            pending.request_messages,
+                            pending.forwarded,
+                            stable_prefix_message_count=max(len(previous_forwarded) - 1, 0),
+                        )
+                        events.append({
+                            'mode': mode,
+                            'session_id': replay.session_id,
+                            'project': replay.decoded_project_path,
+                            'request_id': pending.turn.request_id,
+                            'timestamp': pending.turn.timestamp.isoformat(),
+                            'first_diff_index': idx,
+                            'prev_len': len(previous_forwarded),
+                            'curr_len': len(pending.forwarded),
+                            'prev_msg': previous_forwarded[idx] if idx is not None and idx < len(previous_forwarded) else None,
+                            'curr_msg': pending.forwarded[idx] if idx is not None and idx < len(pending.forwarded) else None,
+                            'prev_tail': previous_forwarded_context[-4:] if previous_forwarded_context else [],
+                            'curr_tail': pending.request_messages[-4:],
+                            'retroactive_rewrite': retro,
+                        })
+                        if len(events) >= max_events_per_mode:
+                            return events
+                previous_forwarded = copy.deepcopy(pending.forwarded)
+                previous_timestamp = pending.turn.timestamp
+            try:
+                prefix_tracker.update_from_response(
+                    cache_read_tokens=0,
+                    cache_write_tokens=0,
+                    messages=forwarded,
+                    message_token_counts=[tokenizer.count_message(msg) for msg in forwarded],
+                    original_messages=conversation,
+                )
+            except TypeError:
+                prefix_tracker.update_from_response(
+                    cache_read_tokens=0,
+                    cache_write_tokens=0,
+                    messages=forwarded,
+                    message_token_counts=[tokenizer.count_message(msg) for msg in forwarded],
+                )
+            class Pending: pass
+            pending = Pending()
+            pending.turn = turn
+            pending.request_messages = copy.deepcopy(conversation)
+            pending.forwarded = forwarded
+            conversation.append(turn.assistant_message)
+            conversation_token_total = raw_input_tokens + tokenizer.count_message(turn.assistant_message)
+            previous_original_context = copy.deepcopy(conversation)
+            previous_forwarded_context = copy.deepcopy(forwarded) + [copy.deepcopy(turn.assistant_message)]
+    return events
+
+print(json.dumps({
+    'token': trace_mode(PROXY_MODE_TOKEN),
+    'cache': trace_mode(PROXY_MODE_CACHE),
+}, indent=2))
+"""
+        env["BUST_TRACE_RECENT"] = str(recent_turns_per_session)
+        env["BUST_TRACE_MAX"] = str(max_events_per_mode)
+        script_path = worktree_dir / "benchmarks" / "claude_session_mode_benchmark.py"
+        if not script_path.exists():
+            script_path = repo_root / "benchmarks" / "claude_session_mode_benchmark.py"
+        env["BUST_TRACE_SCRIPT"] = str(script_path)
+        completed = subprocess.run(
+            [sys.executable, "-c", code],
+            cwd=worktree_dir,
+            check=True,
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+        payload = json.loads(completed.stdout)
+        events: list[BustEvent] = []
+        for mode in ("token", "cache"):
+            for item in payload[mode]:
+                events.append(BustEvent(branch=label, **item))
+        return events
+    except subprocess.CalledProcessError as exc:
+        raise RuntimeError(
+            f"trace failed for {label} ({ref})\nSTDOUT:\n{exc.stdout}\nSTDERR:\n{exc.stderr}"
+        ) from exc
+    finally:
+        subprocess.run(
+            ["git", "worktree", "remove", "--force", str(worktree_dir)],
+            cwd=repo_root,
+            check=True,
+        )
+
+
+def _render_markdown(events: list[BustEvent], recent_turns_per_session: int) -> str:
+    lines = [
+        "# Cache Bust Trace Report",
+        "",
+        f"- Sampling: most recent {recent_turns_per_session} turns per session",
+        "",
+    ]
+    for branch in ("main", "pr"):
+        lines.append(f"## {branch}")
+        lines.append("")
+        branch_events = [e for e in events if e.branch == branch]
+        for mode in ("token", "cache"):
+            lines.append(f"### {mode}")
+            mode_events = [e for e in branch_events if e.mode == mode]
+            if not mode_events:
+                lines.append("")
+                lines.append("- No bust events captured.")
+                lines.append("")
+                continue
+            for event in mode_events:
+                lines.append("")
+                lines.append(
+                    f"- `{event.project}` `{event.session_id}` `{event.request_id}` "
+                    f"{event.timestamp} diff_index={event.first_diff_index} "
+                    f"retroactive={event.retroactive_rewrite}"
+                )
+        lines.append("")
+    return "\n".join(lines)
+
+
+def _render_html(events: list[BustEvent], recent_turns_per_session: int) -> str:
+    sections = []
+    for branch in ("main", "pr"):
+        rows = []
+        branch_events = [e for e in events if e.branch == branch]
+        for mode in ("token", "cache"):
+            mode_events = [e for e in branch_events if e.mode == mode]
+            if not mode_events:
+                rows.append(
+                    f"<tr><td>{mode}</td><td colspan='6'>No bust events captured.</td></tr>"
+                )
+                continue
+            for event in mode_events:
+                rows.append(
+                    "<tr>"
+                    f"<td>{mode}</td>"
+                    f"<td>{event.project}</td>"
+                    f"<td>{event.session_id}</td>"
+                    f"<td>{event.request_id}</td>"
+                    f"<td>{event.timestamp}</td>"
+                    f"<td>{event.first_diff_index}</td>"
+                    f"<td>{event.retroactive_rewrite}</td>"
+                    "</tr>"
+                )
+        sections.append(
+            f"<section><h2>{branch}</h2><table><thead><tr>"
+            "<th>Mode</th><th>Project</th><th>Session</th><th>Request</th>"
+            "<th>Timestamp</th><th>First Diff</th><th>Retroactive</th>"
+            f"</tr></thead><tbody>{''.join(rows)}</tbody></table></section>"
+        )
+    return f"""<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <meta name="viewport" content="width=device-width, initial-scale=1">
+  <title>Cache Bust Trace Report</title>
+  <style>
+    body {{ font-family: 'Segoe UI', system-ui, sans-serif; margin: 0; background: #f8fafc; color: #0f172a; }}
+    .shell {{ max-width: 1280px; margin: 0 auto; padding: 32px 16px 48px; }}
+    h1, h2 {{ letter-spacing: -0.02em; }}
+    section {{ background: white; border: 1px solid #e2e8f0; border-radius: 16px; padding: 20px; margin-top: 16px; }}
+    table {{ width: 100%; border-collapse: collapse; }}
+    th, td {{ padding: 10px 12px; border-bottom: 1px solid #e2e8f0; text-align: left; white-space: nowrap; }}
+    th {{ background: #f1f5f9; }}
+  </style>
+</head>
+<body>
+  <div class="shell">
+    <h1>Cache Bust Trace Report</h1>
+    <p>Most recent {recent_turns_per_session} turns per session.</p>
+    {''.join(sections)}
+  </div>
+</body>
+</html>"""
+
+
+def main() -> int:
+    repo_root = Path(__file__).resolve().parents[1]
+    output_dir = DEFAULT_OUTPUT_DIR
+    output_dir.mkdir(parents=True, exist_ok=True)
+    recent_turns_per_session = 200
+    events = _trace_branch(
+        repo_root, "upstream/main", "main", recent_turns_per_session=recent_turns_per_session
+    )
+    events.extend(
+        _trace_branch(repo_root, "HEAD", "pr", recent_turns_per_session=recent_turns_per_session)
+    )
+
+    md_path = output_dir / "cache_bust_trace.md"
+    json_path = output_dir / "cache_bust_trace.json"
+    html_path = output_dir / "cache_bust_trace.html"
+    md_path.write_text(_render_markdown(events, recent_turns_per_session), encoding="utf-8")
+    json_path.write_text(json.dumps([asdict(event) for event in events], indent=2), encoding="utf-8")
+    html_path.write_text(_render_html(events, recent_turns_per_session), encoding="utf-8")
+    print(md_path)
+    print(json_path)
+    print(html_path)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/cache_validation_bundle.py

@@ -0,0 +1,697 @@
+#!/usr/bin/env python3
+"""Generate a reproducible local cache-validation report bundle."""
+
+from __future__ import annotations
+
+import argparse
+import copy
+import hashlib
+import html
+import json
+import logging
+import platform
+import subprocess
+import sys
+from dataclasses import asdict
+from datetime import timedelta
+from pathlib import Path
+from typing import Any
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import benchmarks.claude_session_mode_benchmark as real_bench
+import benchmarks.synthetic_long_cache_suite_report as long_suite
+import benchmarks.synthetic_token_cache_bust_report as token_bust
+from benchmarks.claude_session_mode_benchmark import (
+    PROXY_MODE_CACHE,
+    PROXY_MODE_TOKEN,
+    _apply_mode_to_messages,
+    _cache_gap_within_ttl,
+    _rewrite_scope,
+    build_dataset_and_observed_from_files,
+    determine_winners,
+    format_currency,
+    get_tokenizer,
+    load_session_replay,
+    resolve_checkpoint_dir,
+    select_session_files,
+    simulate_session_files,
+    trim_replay_to_recent_turns,
+    write_report,
+)
+from headroom.cache.compression_cache import CompressionCache
+from headroom.cache.prefix_tracker import PrefixCacheTracker
+
+DEFAULT_OUTPUT_DIR = Path("benchmark_results") / "cache_validation_bundle"
+
+
+def _excerpt_content(content: Any, *, max_chars: int) -> str:
+    if isinstance(content, str):
+        text = content.replace("\n", " ")
+        return text[:max_chars] + ("..." if len(text) > max_chars else "")
+    if isinstance(content, list):
+        parts = []
+        for block in content[:4]:
+            if isinstance(block, dict):
+                btype = str(block.get("type", "unknown"))
+                bcontent = block.get("content", "")
+                if isinstance(bcontent, str):
+                    bcontent = bcontent.replace("\n", " ")
+                    bcontent = bcontent[:max_chars] + ("..." if len(bcontent) > max_chars else "")
+                parts.append(f"[{btype}] {bcontent}")
+            else:
+                parts.append(str(block)[:max_chars])
+        return " | ".join(parts)
+    return str(content)[:max_chars]
+
+
+def _message_preview(msg: dict[str, Any], *, max_chars: int) -> dict[str, str]:
+    return {
+        "role": str(msg.get("role")),
+        "content_excerpt": _excerpt_content(msg.get("content"), max_chars=max_chars),
+    }
+
+
+def _stable_hash(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:12]
+
+
+def _redact_text(value: str, *, prefix: str) -> str:
+    return f"{prefix}-{_stable_hash(value)}"
+
+
+def _redact_path(value: str) -> str:
+    path = Path(value)
+    suffix = path.suffix
+    return f"path-{_stable_hash(value)}{suffix}"
+
+
+def _git_output(args: list[str], cwd: Path) -> str | None:
+    try:
+        completed = subprocess.run(
+            ["git", *args],
+            cwd=cwd,
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        return completed.stdout.strip()
+    except Exception:
+        return None
+
+
+def _runtime_metadata(repo_root: Path) -> dict[str, Any]:
+    return {
+        "git_sha": _git_output(["rev-parse", "HEAD"], repo_root),
+        "git_dirty": bool(_git_output(["status", "--porcelain"], repo_root)),
+        "python_version": sys.version,
+        "platform": platform.platform(),
+        "implementation": platform.python_implementation(),
+    }
+
+
+def _corpus_fingerprint(
+    *,
+    root: Path,
+    session_files: list[Path],
+    max_sessions: int | None,
+    recent_turns_per_session: int | None,
+    cache_ttl_minutes: int,
+) -> dict[str, Any]:
+    normalized_files = [str(p.resolve()) for p in session_files]
+    payload = {
+        "root": str(root.resolve()),
+        "session_files": normalized_files,
+        "max_sessions": max_sessions,
+        "recent_turns_per_session": recent_turns_per_session,
+        "cache_ttl_minutes": cache_ttl_minutes,
+    }
+    digest = hashlib.sha256(json.dumps(payload, sort_keys=True).encode("utf-8")).hexdigest()
+    return {
+        "root": str(root.resolve()),
+        "session_file_count": len(session_files),
+        "session_files_sha256": digest,
+        "max_sessions": max_sessions,
+        "recent_turns_per_session": recent_turns_per_session,
+        "cache_ttl_minutes": cache_ttl_minutes,
+    }
+
+
+def _collect_real_processed_events(
+    *,
+    root: Path,
+    recent_turns_per_session: int | None,
+    max_events_per_mode: int,
+    ttl_minutes: int,
+    max_chars: int,
+    include_content: bool,
+) -> dict[str, Any]:
+    ttl = timedelta(minutes=ttl_minutes)
+    events: list[dict[str, Any]] = []
+    session_files = select_session_files(root)
+    for mode in (PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        proxy = real_bench._make_proxy(mode)
+        collected = 0
+        for session_file in session_files:
+            replay = load_session_replay(session_file)
+            if replay is None:
+                continue
+            replay = trim_replay_to_recent_turns(replay, recent_turns_per_session)
+            prefix_tracker = PrefixCacheTracker("anthropic")
+            comp_cache = CompressionCache() if mode == PROXY_MODE_TOKEN else None
+            conversation: list[dict[str, Any]] = []
+            previous_original_context: list[dict[str, Any]] | None = None
+            previous_forwarded_context: list[dict[str, Any]] | None = None
+            previous_forwarded: list[dict[str, Any]] = []
+            previous_timestamp = None
+            pending = None
+            for turn in replay.turns:
+                tokenizer = get_tokenizer(turn.model)
+                prior_context_message_count = len(conversation)
+                conversation.extend(turn.input_messages)
+                forwarded = _apply_mode_to_messages(
+                    proxy,
+                    mode,
+                    conversation,
+                    model=turn.model,
+                    prefix_tracker=prefix_tracker,
+                    comp_cache=comp_cache,
+                    previous_original_messages=previous_original_context,
+                    previous_forwarded_messages=previous_forwarded_context,
+                )
+                rewrite, retro = _rewrite_scope(
+                    conversation,
+                    forwarded,
+                    stable_prefix_message_count=prior_context_message_count,
+                )
+                if rewrite:
+                    prior_forwarded = pending.forwarded if pending is not None else previous_forwarded
+                    prior_ts = pending.turn.timestamp if pending is not None else previous_timestamp
+                    eligible = bool(
+                        prior_ts is not None
+                        and _cache_gap_within_ttl(turn.timestamp, prior_ts, ttl=ttl)
+                        and prior_forwarded
+                    )
+                    prefix_preserved = None
+                    first_diff_index = None
+                    if eligible:
+                        prefix_preserved = (
+                            len(forwarded) >= len(prior_forwarded)
+                            and forwarded[: len(prior_forwarded)] == prior_forwarded
+                        )
+                        if not prefix_preserved:
+                            for idx, (a, b) in enumerate(zip(prior_forwarded, forwarded)):
+                                if a != b:
+                                    first_diff_index = idx
+                                    break
+                            if first_diff_index is None:
+                                first_diff_index = min(len(prior_forwarded), len(forwarded))
+                    events.append(
+                        {
+                            "mode": mode,
+                            "session_id": replay.session_id if include_content else _redact_text(replay.session_id, prefix="session"),
+                            "project": replay.decoded_project_path if include_content else _redact_path(replay.decoded_project_path),
+                            "request_id": turn.request_id if include_content else _redact_text(turn.request_id, prefix="request"),
+                            "timestamp": turn.timestamp.isoformat(),
+                            "cache_eligible": eligible,
+                            "prefix_preserved": prefix_preserved,
+                            "retroactive_rewrite": retro,
+                            "first_diff_index": first_diff_index,
+                            "original_tail": [
+                                _message_preview(m, max_chars=max_chars) if include_content else {
+                                    "role": str(m.get("role")),
+                                    "content_excerpt": "[redacted]",
+                                }
+                                for m in conversation[max(0, len(conversation) - 4) :]
+                            ],
+                            "forwarded_tail": [
+                                _message_preview(m, max_chars=max_chars) if include_content else {
+                                    "role": str(m.get("role")),
+                                    "content_excerpt": "[redacted]",
+                                }
+                                for m in forwarded[max(0, len(forwarded) - 4) :]
+                            ],
+                        }
+                    )
+                    collected += 1
+                    if collected >= max_events_per_mode:
+                        break
+                if pending is not None:
+                    previous_forwarded = copy.deepcopy(pending.forwarded)
+                    previous_timestamp = pending.turn.timestamp
+                real_bench._update_prefix_tracker(
+                    prefix_tracker,
+                    cache_read_tokens=0,
+                    cache_write_tokens=0,
+                    messages=forwarded,
+                    message_token_counts=[tokenizer.count_message(msg) for msg in forwarded],
+                    original_messages=conversation,
+                )
+                class Pending:
+                    pass
+
+                pending = Pending()
+                pending.turn = turn
+                pending.forwarded = forwarded
+                conversation.append(turn.assistant_message)
+                previous_original_context = copy.deepcopy(conversation)
+                previous_forwarded_context = copy.deepcopy(forwarded) + [
+                    copy.deepcopy(turn.assistant_message)
+                ]
+            if collected >= max_events_per_mode:
+                break
+    return {"events": events}
+
+
+def _write_processed_event_reports(output_dir: Path, payload: dict[str, Any]) -> tuple[Path, Path, Path]:
+    out_dir = output_dir / "real_processed"
+    out_dir.mkdir(parents=True, exist_ok=True)
+    json_path = out_dir / "real_processed_rewrite_report.json"
+    md_path = out_dir / "real_processed_rewrite_report.md"
+    html_path = out_dir / "real_processed_rewrite_report.html"
+    json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+    md = [
+        "# Real Processed Rewrite Report",
+        "",
+        "Local-only report from real Claude transcript replays. Do not commit.",
+        "",
+    ]
+    for mode in (PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        mode_events = [e for e in payload["events"] if e["mode"] == mode]
+        md.extend([f"## `{mode}`", ""])
+        if not mode_events:
+            md.extend(["No rewrite events captured.", ""])
+            continue
+        for i, e in enumerate(mode_events, start=1):
+            md.extend(
+                [
+                    f"### Event {i}",
+                    "",
+                    f"- session: `{e['session_id']}`",
+                    f"- request: `{e['request_id']}`",
+                    f"- cache eligible: `{e['cache_eligible']}`",
+                    f"- prefix preserved: `{e['prefix_preserved']}`",
+                    f"- retroactive rewrite: `{e['retroactive_rewrite']}`",
+                    f"- first diff index: `{e['first_diff_index']}`",
+                    "",
+                    "**Original Tail**",
+                    "",
+                ]
+            )
+            for msg in e["original_tail"]:
+                md.append(f"- `{msg['role']}`: {msg['content_excerpt']}")
+            md.extend(["", "**Forwarded Tail**", ""])
+            for msg in e["forwarded_tail"]:
+                md.append(f"- `{msg['role']}`: {msg['content_excerpt']}")
+            md.extend(["", ""])
+    md_path.write_text("\n".join(md), encoding="utf-8")
+
+    sections = []
+    for mode in (PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        mode_events = [e for e in payload["events"] if e["mode"] == mode]
+        cards = []
+        for i, e in enumerate(mode_events, start=1):
+            orig = "".join(
+                f"<li><code>{html.escape(str(m['role']))}</code>: "
+                f"{html.escape(str(m['content_excerpt']))}</li>"
+                for m in e["original_tail"]
+            )
+            fwd = "".join(
+                f"<li><code>{html.escape(str(m['role']))}</code>: "
+                f"{html.escape(str(m['content_excerpt']))}</li>"
+                for m in e["forwarded_tail"]
+            )
+            cards.append(
+                "<div class='event'>"
+                f"<h3>Event {i}</h3>"
+                f"<p><strong>session</strong>: <code>{html.escape(e['session_id'])}</code><br>"
+                f"<strong>request</strong>: <code>{html.escape(e['request_id'])}</code><br>"
+                f"<strong>cache eligible</strong>: <code>{e['cache_eligible']}</code><br>"
+                f"<strong>prefix preserved</strong>: <code>{e['prefix_preserved']}</code><br>"
+                f"<strong>retroactive rewrite</strong>: <code>{e['retroactive_rewrite']}</code><br>"
+                f"<strong>first diff index</strong>: <code>{e['first_diff_index']}</code></p>"
+                f"<div class='cols'><div><h4>Original Tail</h4><ul>{orig}</ul></div>"
+                f"<div><h4>Forwarded Tail</h4><ul>{fwd}</ul></div></div>"
+                "</div>"
+            )
+        sections.append(
+            f"<section class='card'><h2>{html.escape(mode)}</h2>"
+            + ("".join(cards) if cards else "<p>No rewrite events captured.</p>")
+            + "</section>"
+        )
+
+    html_doc = (
+        "<!doctype html><html><head><meta charset='utf-8'>"
+        "<meta name='viewport' content='width=device-width, initial-scale=1'>"
+        "<title>Real Processed Rewrite Report</title>"
+        "<style>"
+        "body{font-family:ui-sans-serif,system-ui,sans-serif;max-width:1200px;margin:40px auto;padding:0 20px;line-height:1.55;color:#111827;background:#f8fafc}"
+        ".card,.event{background:white;border:1px solid #cbd5e1;border-radius:16px;padding:20px;margin:18px 0;box-shadow:0 8px 24px rgba(15,23,42,.06)}"
+        ".cols{display:grid;grid-template-columns:1fr 1fr;gap:20px} code{background:#e5e7eb;padding:1px 4px;border-radius:4px} ul{padding-left:20px}"
+        "</style></head><body>"
+        "<h1>Real Processed Rewrite Report</h1>"
+        "<div class='card'><p>Local-only report from real Claude transcript replays. Do not commit.</p></div>"
+        + "".join(sections)
+        + "</body></html>"
+    )
+    html_path.write_text(html_doc, encoding="utf-8")
+    return md_path, json_path, html_path
+
+
+def _write_index(
+    output_dir: Path,
+    *,
+    args: argparse.Namespace,
+    dataset: dict[str, Any],
+    observed: dict[str, Any],
+    summaries: dict[str, Any],
+    winners: dict[str, str],
+    metadata: dict[str, Any],
+    corpus: dict[str, Any],
+    processed_paths: tuple[Path, Path, Path],
+    token_bust_paths: tuple[Path, Path, Path],
+    long_suite_paths: tuple[Path, Path, Path],
+) -> tuple[Path, Path]:
+    md_path = output_dir / "index.md"
+    html_path = output_dir / "index.html"
+    md_lines = [
+        "# Cache Validation Bundle",
+        "",
+        "This bundle is reproducible on another machine with local Claude transcript data in `~/.claude/projects`.",
+        "",
+        "## Configuration",
+        "",
+        f"- root: `{args.root}`",
+        f"- output dir: `{args.output_dir}`",
+        f"- recent turns per session: `{args.recent_turns_per_session}`",
+        f"- workers: `{args.workers}`",
+        f"- cache TTL minutes: `{args.cache_ttl_minutes}`",
+        f"- cache write multiplier: `{args.cache_write_multiplier}`",
+        f"- max real processed events per mode: `{args.max_real_events_per_mode}`",
+        f"- include transcript content: `{args.include_content}`",
+        "",
+        "## Reproducibility",
+        "",
+        f"- git sha: `{metadata['git_sha']}`",
+        f"- git dirty: `{metadata['git_dirty']}`",
+        f"- python: `{metadata['implementation']}`",
+        f"- platform: `{metadata['platform']}`",
+        f"- corpus session file count: `{corpus['session_file_count']}`",
+        f"- corpus fingerprint: `{corpus['session_files_sha256']}`",
+        "",
+        "## Real Corpus Summary",
+        "",
+        f"- projects: `{dataset['projects']}`",
+        f"- sessions: `{dataset['sessions']}`",
+        f"- requests: `{dataset['requests']}`",
+        f"- observed total cost: `{format_currency(observed['total_cost_usd'])}`",
+        f"- winner by total cost: `{winners['total_cost']}`",
+        "",
+        "| Mode | Total Cost | Cache Busts | Busting Rewrites | Stable Replay Rewrites | Rewrites | Retroactive Rewrites | TTL Expiry | Forwarded Tokens |",
+        "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = summaries[mode]
+        md_lines.append(
+            f"| `{mode}` | {format_currency(summary['total_cost_usd'])} | {summary['cache_bust_turns']} | "
+            f"{summary['busting_rewrite_turns']} | {summary['stable_replay_rewrite_turns']} | "
+            f"{summary['rewrite_turns']} | {summary['retroactive_rewrite_turns']} | "
+            f"{summary['ttl_expiry_turns']} | {summary['forwarded_input_tokens']:,} |"
+        )
+    md_lines.extend(
+        [
+            "",
+            "## Interpretation",
+            "",
+            "- `cache_bust_turns` and `busting_rewrite_turns` are the hard-failure metrics for Anthropic prefix caching.",
+            "- `stable_replay_rewrite_turns` indicates replay of previously-forwarded bytes that still preserves cache prefix stability.",
+            "- `retroactive_rewrite_turns` is descriptive only; it does not imply a cache break by itself.",
+            "- `ttl_expiry_turns` is workload timing context, not compression correctness.",
+            "",
+            "## Artifacts",
+            "",
+            f"- real corpus summary markdown: [real/{real_bench.OUTPUT_MD}](real/{real_bench.OUTPUT_MD})",
+            f"- real corpus summary html: [real/{real_bench.OUTPUT_HTML}](real/{real_bench.OUTPUT_HTML})",
+            f"- real processed markdown: [real_processed/{processed_paths[0].name}](real_processed/{processed_paths[0].name})",
+            f"- real processed html: [real_processed/{processed_paths[2].name}](real_processed/{processed_paths[2].name})",
+            f"- synthetic token bust markdown: [synthetic_token_bust/{token_bust_paths[0].name}](synthetic_token_bust/{token_bust_paths[0].name})",
+            f"- synthetic token bust html: [synthetic_token_bust/{token_bust_paths[2].name}](synthetic_token_bust/{token_bust_paths[2].name})",
+            f"- synthetic long suite markdown: [synthetic_long_suite/{long_suite_paths[0].name}](synthetic_long_suite/{long_suite_paths[0].name})",
+            f"- synthetic long suite html: [synthetic_long_suite/{long_suite_paths[2].name}](synthetic_long_suite/{long_suite_paths[2].name})",
+        ]
+    )
+    md_path.write_text("\n".join(md_lines), encoding="utf-8")
+
+    rows = []
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = summaries[mode]
+        rows.append(
+            "<tr>"
+            f"<td><code>{html.escape(mode)}</code></td>"
+            f"<td>{html.escape(format_currency(summary['total_cost_usd']))}</td>"
+            f"<td>{summary['cache_bust_turns']}</td>"
+            f"<td>{summary['busting_rewrite_turns']}</td>"
+            f"<td>{summary['stable_replay_rewrite_turns']}</td>"
+            f"<td>{summary['rewrite_turns']}</td>"
+            f"<td>{summary['retroactive_rewrite_turns']}</td>"
+            f"<td>{summary['ttl_expiry_turns']}</td>"
+            f"<td>{summary['forwarded_input_tokens']:,}</td>"
+            "</tr>"
+        )
+    html_doc = (
+        "<!doctype html><html><head><meta charset='utf-8'>"
+        "<meta name='viewport' content='width=device-width, initial-scale=1'>"
+        "<title>Cache Validation Bundle</title>"
+        "<style>"
+        "body{font-family:ui-sans-serif,system-ui,sans-serif;max-width:1200px;margin:40px auto;padding:0 20px;line-height:1.55;color:#111827;background:#f8fafc}"
+        ".card{background:white;border:1px solid #cbd5e1;border-radius:16px;padding:24px;margin:18px 0;box-shadow:0 8px 24px rgba(15,23,42,.06)}"
+        "table{border-collapse:collapse;width:100%;margin:16px 0;background:white}"
+        "th,td{border:1px solid #cbd5e1;padding:10px;text-align:left}th{background:#e2e8f0}"
+        "code{background:#e5e7eb;padding:1px 4px;border-radius:4px}"
+        "</style></head><body>"
+        "<h1>Cache Validation Bundle</h1>"
+        "<div class='card'>"
+        f"<p><strong>root</strong>: <code>{html.escape(str(args.root))}</code><br>"
+        f"<strong>recent turns per session</strong>: <code>{html.escape(str(args.recent_turns_per_session))}</code><br>"
+        f"<strong>workers</strong>: <code>{args.workers}</code><br>"
+        f"<strong>cache TTL minutes</strong>: <code>{args.cache_ttl_minutes}</code><br>"
+        f"<strong>include transcript content</strong>: <code>{args.include_content}</code></p>"
+        "</div>"
+        "<div class='card'><h2>Reproducibility</h2>"
+        f"<p><strong>git sha</strong>: <code>{html.escape(str(metadata['git_sha']))}</code><br>"
+        f"<strong>git dirty</strong>: <code>{metadata['git_dirty']}</code><br>"
+        f"<strong>python</strong>: <code>{html.escape(str(metadata['implementation']))}</code><br>"
+        f"<strong>platform</strong>: <code>{html.escape(str(metadata['platform']))}</code><br>"
+        f"<strong>corpus session file count</strong>: <code>{corpus['session_file_count']}</code><br>"
+        f"<strong>corpus fingerprint</strong>: <code>{html.escape(str(corpus['session_files_sha256']))}</code></p>"
+        "</div>"
+        "<div class='card'><h2>Real Corpus Summary</h2>"
+        f"<p>projects: <code>{dataset['projects']}</code><br>"
+        f"sessions: <code>{dataset['sessions']}</code><br>"
+        f"requests: <code>{dataset['requests']}</code><br>"
+        f"observed total cost: <code>{html.escape(format_currency(observed['total_cost_usd']))}</code><br>"
+        f"winner by total cost: <code>{html.escape(winners['total_cost'])}</code></p>"
+        "<table><thead><tr><th>Mode</th><th>Total Cost</th><th>Cache Busts</th><th>Busting Rewrites</th>"
+        "<th>Stable Replay Rewrites</th><th>Rewrites</th>"
+        "<th>Retroactive Rewrites</th><th>TTL Expiry</th><th>Forwarded Tokens</th></tr></thead><tbody>"
+        + "".join(rows)
+        + "</tbody></table>"
+        "<p><strong>Interpretation</strong>: <code>cache_bust_turns</code> and "
+        "<code>busting_rewrite_turns</code> are the hard-failure metrics. "
+        "<code>stable_replay_rewrite_turns</code> is acceptable stable replay. "
+        "<code>retroactive_rewrite_turns</code> is descriptive only. "
+        "<code>ttl_expiry_turns</code> is workload timing context.</p></div>"
+        "<div class='card'><h2>Artifacts</h2><ul>"
+        f"<li><a href='real/{real_bench.OUTPUT_HTML}'>Real corpus summary HTML</a></li>"
+        f"<li><a href='real/{real_bench.OUTPUT_MD}'>Real corpus summary Markdown</a></li>"
+        f"<li><a href='real_processed/{processed_paths[2].name}'>Real processed rewrite HTML</a></li>"
+        f"<li><a href='real_processed/{processed_paths[0].name}'>Real processed rewrite Markdown</a></li>"
+        f"<li><a href='synthetic_token_bust/{token_bust_paths[2].name}'>Synthetic token-bust HTML</a></li>"
+        f"<li><a href='synthetic_token_bust/{token_bust_paths[0].name}'>Synthetic token-bust Markdown</a></li>"
+        f"<li><a href='synthetic_long_suite/{long_suite_paths[2].name}'>Synthetic long suite HTML</a></li>"
+        f"<li><a href='synthetic_long_suite/{long_suite_paths[0].name}'>Synthetic long suite Markdown</a></li>"
+        "</ul></div></body></html>"
+    )
+    html_path.write_text(html_doc, encoding="utf-8")
+    return md_path, html_path
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, default=real_bench.DEFAULT_ROOT)
+    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
+    parser.add_argument("--recent-turns-per-session", type=int, default=None)
+    parser.add_argument("--workers", type=int, default=1)
+    parser.add_argument("--cache-ttl-minutes", type=int, default=real_bench.DEFAULT_CACHE_TTL_MINUTES)
+    parser.add_argument("--cache-write-multiplier", type=float, default=1.25)
+    parser.add_argument("--max-sessions", type=int, default=None)
+    parser.add_argument("--max-real-events-per-mode", type=int, default=8)
+    parser.add_argument("--content-excerpt-chars", type=int, default=220)
+    parser.add_argument(
+        "--include-content",
+        action="store_true",
+        help="Include real transcript-derived content excerpts in the processed event reports.",
+    )
+    parser.add_argument(
+        "--checkpoint-dir",
+        type=Path,
+        default=real_bench.DEFAULT_OUTPUT_DIR / real_bench.CHECKPOINT_DIRNAME,
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    output_dir = args.output_dir
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    logging.getLogger("headroom.transforms").setLevel(logging.WARNING)
+    logging.getLogger("headroom.proxy").setLevel(logging.WARNING)
+
+    session_files = select_session_files(args.root, max_sessions=args.max_sessions)
+    if not session_files:
+        print(f"No Claude session replays found under {args.root}")
+        return 1
+
+    repo_root = Path(__file__).resolve().parents[1]
+    metadata = _runtime_metadata(repo_root)
+    corpus = _corpus_fingerprint(
+        root=args.root,
+        session_files=session_files,
+        max_sessions=args.max_sessions,
+        recent_turns_per_session=args.recent_turns_per_session,
+        cache_ttl_minutes=args.cache_ttl_minutes,
+    )
+    dataset, observed = build_dataset_and_observed_from_files(
+        session_files,
+        cache_write_multiplier=args.cache_write_multiplier,
+        recent_turns_per_session=args.recent_turns_per_session,
+    )
+    checkpoint_base = output_dir / "checkpoints" / corpus["session_files_sha256"]
+    checkpoint_dir = resolve_checkpoint_dir(
+        checkpoint_base,
+        recent_turns_per_session=args.recent_turns_per_session,
+        cache_ttl_minutes=args.cache_ttl_minutes,
+    )
+
+    real_output_dir = output_dir / "real"
+    summaries = simulate_session_files(
+        session_files,
+        dataset,
+        cache_ttl_minutes=args.cache_ttl_minutes,
+        cache_write_multiplier=args.cache_write_multiplier,
+        workers=args.workers,
+        checkpoint_dir=checkpoint_dir,
+        recent_turns_per_session=args.recent_turns_per_session,
+    )
+    real_md, real_json, real_html = write_report(real_output_dir, dataset, observed, summaries)
+
+    processed_payload = _collect_real_processed_events(
+        root=args.root,
+        recent_turns_per_session=args.recent_turns_per_session,
+        max_events_per_mode=args.max_real_events_per_mode,
+        ttl_minutes=args.cache_ttl_minutes,
+        max_chars=args.content_excerpt_chars,
+        include_content=args.include_content,
+    )
+    processed_paths = _write_processed_event_reports(output_dir, processed_payload)
+
+    token_bust.OUTPUT_DIR = output_dir / "synthetic_token_bust"
+    token_bust_replay = token_bust._build_replay()
+    original_make_proxy = token_bust.bench._make_proxy
+    token_bust.bench._make_proxy = lambda mode: token_bust._FakeProxy()
+    try:
+        _, token_bust_summaries = token_bust.simulate_replays(
+            [token_bust_replay], cache_ttl_minutes=token_bust.TTL_MINUTES if hasattr(token_bust, "TTL_MINUTES") else 5
+        )
+        token_bust_events = token_bust._build_bust_events(token_bust_replay)
+    finally:
+        token_bust.bench._make_proxy = original_make_proxy
+    token_bust_paths = token_bust._write_report(
+        token_bust_replay,
+        token_bust_summaries,
+        determine_winners(token_bust_summaries),
+        token_bust_events,
+    )
+
+    long_suite.OUTPUT_DIR = output_dir / "synthetic_long_suite"
+    per_scenario, aggregate = long_suite._run_suite()
+    long_suite_paths = long_suite._write_report(per_scenario, aggregate)
+
+    bundle_payload = {
+        "config": {
+            "root": _redact_path(str(args.root.resolve())),
+            "output_dir": _redact_path(str(output_dir.resolve())),
+            "recent_turns_per_session": args.recent_turns_per_session,
+            "workers": args.workers,
+            "cache_ttl_minutes": args.cache_ttl_minutes,
+            "cache_write_multiplier": args.cache_write_multiplier,
+            "max_sessions": args.max_sessions,
+            "max_real_events_per_mode": args.max_real_events_per_mode,
+            "content_excerpt_chars": args.content_excerpt_chars,
+            "include_content": args.include_content,
+            "checkpoint_dir": _redact_path(str(checkpoint_dir.resolve())),
+        },
+        "runtime": metadata,
+        "corpus": corpus,
+        "real": {
+            "dataset": asdict(dataset),
+            "observed": asdict(observed),
+            "summaries": {mode: asdict(summary) for mode, summary in summaries.items()},
+            "winners": determine_winners(summaries),
+            "paths": {
+                "markdown": str(real_md),
+                "json": str(real_json),
+                "html": str(real_html),
+            },
+        },
+        "processed_real": {
+            "events": processed_payload["events"],
+            "paths": {
+                "markdown": str(processed_paths[0]),
+                "json": str(processed_paths[1]),
+                "html": str(processed_paths[2]),
+            },
+        },
+        "synthetic_token_bust": {
+            "paths": {
+                "markdown": str(token_bust_paths[0]),
+                "json": str(token_bust_paths[1]),
+                "html": str(token_bust_paths[2]),
+            }
+        },
+        "synthetic_long_suite": {
+            "paths": {
+                "markdown": str(long_suite_paths[0]),
+                "json": str(long_suite_paths[1]),
+                "html": str(long_suite_paths[2]),
+            }
+        },
+    }
+    manifest_path = output_dir / "bundle_manifest.json"
+    manifest_path.write_text(json.dumps(bundle_payload, indent=2), encoding="utf-8")
+
+    index_md, index_html = _write_index(
+        output_dir,
+        args=args,
+        dataset=asdict(dataset),
+        observed=asdict(observed),
+        summaries={mode: asdict(summary) for mode, summary in summaries.items()},
+        winners=determine_winners(summaries),
+        metadata=metadata,
+        corpus=corpus,
+        processed_paths=processed_paths,
+        token_bust_paths=token_bust_paths,
+        long_suite_paths=long_suite_paths,
+    )
+
+    print(f"Index markdown: {index_md}")
+    print(f"Index html: {index_html}")
+    print(f"Manifest: {manifest_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/claude_session_mode_benchmark.py

@@ -102,6 +102,9 @@ class ModeSummary:
     cache_bust_turns: int = 0
     ttl_expiry_turns: int = 0
     rewrite_turns: int = 0
+    stable_replay_rewrite_turns: int = 0
+    busting_rewrite_turns: int = 0
+    non_cache_eligible_rewrite_turns: int = 0
     retroactive_rewrite_turns: int = 0
     latest_turn_only_rewrite_turns: int = 0
     turns: list[TurnMetrics] = field(default_factory=list)
@@ -157,6 +160,9 @@ IMPACT_DIRECTION = {
     "cache_bust_turns": "lower",
     "ttl_expiry_turns": "lower",
     "rewrite_turns": "lower",
+    "stable_replay_rewrite_turns": "lower",
+    "busting_rewrite_turns": "lower",
+    "non_cache_eligible_rewrite_turns": "lower",
     "retroactive_rewrite_turns": "lower",
     "latest_turn_only_rewrite_turns": "lower",
 }
@@ -248,6 +254,9 @@ def _mode_summary_from_dict(data: dict[str, Any]) -> ModeSummary:
         cache_bust_turns=data.get("cache_bust_turns", 0),
         ttl_expiry_turns=data.get("ttl_expiry_turns", 0),
         rewrite_turns=data.get("rewrite_turns", 0),
+        stable_replay_rewrite_turns=data.get("stable_replay_rewrite_turns", 0),
+        busting_rewrite_turns=data.get("busting_rewrite_turns", 0),
+        non_cache_eligible_rewrite_turns=data.get("non_cache_eligible_rewrite_turns", 0),
         retroactive_rewrite_turns=data.get("retroactive_rewrite_turns", 0),
         latest_turn_only_rewrite_turns=data.get("latest_turn_only_rewrite_turns", 0),
         turns=turns,
@@ -462,7 +471,7 @@ def resolve_checkpoint_dir(
     recent_turns_per_session: int | None = None,
     cache_ttl_minutes: int = DEFAULT_CACHE_TTL_MINUTES,
 ) -> Path:
-    suffix_parts = ["v4", f"ttl_{cache_ttl_minutes}m"]
+    suffix_parts = ["v5", f"ttl_{cache_ttl_minutes}m"]
     if recent_turns_per_session:
         suffix_parts.append(f"recent_{recent_turns_per_session}")
     else:
@@ -906,6 +915,8 @@ class _PendingTurn:
     raw_input_tokens: int
     request_messages: list[dict[str, Any]]
     forwarded: list[dict[str, Any]]
+    rewrite: bool
+    retroactive_rewrite: bool
 
 
 def _cache_gap_within_ttl(
@@ -1021,6 +1032,9 @@ def _merge_mode_summary(target: ModeSummary, source: ModeSummary) -> None:
     target.cache_bust_turns += source.cache_bust_turns
     target.ttl_expiry_turns += source.ttl_expiry_turns
     target.rewrite_turns += source.rewrite_turns
+    target.stable_replay_rewrite_turns += source.stable_replay_rewrite_turns
+    target.busting_rewrite_turns += source.busting_rewrite_turns
+    target.non_cache_eligible_rewrite_turns += source.non_cache_eligible_rewrite_turns
     target.retroactive_rewrite_turns += source.retroactive_rewrite_turns
     target.latest_turn_only_rewrite_turns += source.latest_turn_only_rewrite_turns
 
@@ -1169,6 +1183,25 @@ def _simulate_single_replay_mode(
                 summary.retroactive_rewrite_turns += 1
             else:
                 summary.latest_turn_only_rewrite_turns += 1
+            prior_forwarded_for_rewrite = pending.forwarded if pending is not None else previous_forwarded
+            prior_timestamp_for_rewrite = (
+                pending.turn.timestamp if pending is not None else previous_timestamp
+            )
+            if (
+                prior_timestamp_for_rewrite is not None
+                and _cache_gap_within_ttl(turn.timestamp, prior_timestamp_for_rewrite, ttl=ttl)
+                and prior_forwarded_for_rewrite
+            ):
+                prefix_preserved = (
+                    len(forwarded) >= len(prior_forwarded_for_rewrite)
+                    and forwarded[: len(prior_forwarded_for_rewrite)] == prior_forwarded_for_rewrite
+                )
+                if prefix_preserved:
+                    summary.stable_replay_rewrite_turns += 1
+                else:
+                    summary.busting_rewrite_turns += 1
+            else:
+                summary.non_cache_eligible_rewrite_turns += 1
         if pending is not None:
             _apply_turn_metrics(
                 pending.summary,
@@ -1203,6 +1236,8 @@ def _simulate_single_replay_mode(
             raw_input_tokens=raw_input_tokens,
             request_messages=copy.deepcopy(conversation),
             forwarded=forwarded,
+            rewrite=rewrite,
+            retroactive_rewrite=retroactive_rewrite,
         )
         conversation.append(turn.assistant_message)
         conversation_token_total = raw_input_tokens + tokenizer.count_message(
@@ -1513,7 +1548,7 @@ def print_console_report(dataset: DatasetSummary, summaries: dict[str, ModeSumma
     print(f"Sampling: {dataset.sampling_note}")
     print()
     print(
-        "mode      raw_tok      cache_tok    cache_read   cache_write   paid_in      paid_out     busts   ttl_exp   rewrite   retro_rw   total_cost    no_cache"
+        "mode      raw_tok      cache_tok    cache_read   cache_write   paid_in      paid_out     busts   ttl_exp   rewrite   stable_rw  bust_rw   noncache_rw  retro_rw   total_cost    no_cache"
     )
     for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
         summary = summaries[mode]
@@ -1522,7 +1557,9 @@ def print_console_report(dataset: DatasetSummary, summaries: dict[str, ModeSumma
             f"{summary.cache_read_tokens:>11,} {summary.cache_write_tokens:>12,} "
             f"{summary.regular_input_tokens:>10,} {summary.output_tokens:>12,} "
             f"{summary.cache_bust_turns:>7,} {summary.ttl_expiry_turns:>9,} "
-            f"{summary.rewrite_turns:>9,} {summary.retroactive_rewrite_turns:>10,} "
+            f"{summary.rewrite_turns:>9,} {summary.stable_replay_rewrite_turns:>10,} "
+            f"{summary.busting_rewrite_turns:>8,} {summary.non_cache_eligible_rewrite_turns:>12,} "
+            f"{summary.retroactive_rewrite_turns:>10,} "
             f"{format_currency(summary.total_cost_usd):>11} "
             f"{format_currency(summary.no_cache_total_cost_usd):>11}"
         )
@@ -1549,6 +1586,12 @@ def print_console_report(dataset: DatasetSummary, summaries: dict[str, ModeSumma
             f"({int(impact['regular_input_tokens']['delta']):,}), "
             f"rewrite={impact['rewrite_turns']['impact']} "
             f"({int(impact['rewrite_turns']['delta']):,}), "
+            f"stable_rw={impact['stable_replay_rewrite_turns']['impact']} "
+            f"({int(impact['stable_replay_rewrite_turns']['delta']):,}), "
+            f"bust_rw={impact['busting_rewrite_turns']['impact']} "
+            f"({int(impact['busting_rewrite_turns']['delta']):,}), "
+            f"noncache_rw={impact['non_cache_eligible_rewrite_turns']['impact']} "
+            f"({int(impact['non_cache_eligible_rewrite_turns']['delta']):,}), "
             f"retro_rw={impact['retroactive_rewrite_turns']['impact']} "
             f"({int(impact['retroactive_rewrite_turns']['delta']):,}), "
             f"window={impact['prompt_window_with_cache']['impact']} "
@@ -1602,6 +1645,9 @@ def build_report_markdown(
                     f"{summary.cache_bust_turns:,}",
                     f"{summary.ttl_expiry_turns:,}",
                     f"{summary.rewrite_turns:,}",
+                    f"{summary.stable_replay_rewrite_turns:,}",
+                    f"{summary.busting_rewrite_turns:,}",
+                    f"{summary.non_cache_eligible_rewrite_turns:,}",
                     f"{summary.retroactive_rewrite_turns:,}",
                     f"{summary.latest_turn_only_rewrite_turns:,}",
                     f"{summary.prompt_window_with_cache:,}",
@@ -1622,6 +1668,9 @@ def build_report_markdown(
             ("prompt_window_without_cache_reads", "Window Without Cache Reads"),
             ("cache_bust_turns", "Cache Bust Turns"),
             ("rewrite_turns", "Rewrite Turns"),
+            ("stable_replay_rewrite_turns", "Stable Replay Rewrite Turns"),
+            ("busting_rewrite_turns", "Busting Rewrite Turns"),
+            ("non_cache_eligible_rewrite_turns", "Non-Cache-Eligible Rewrite Turns"),
             ("retroactive_rewrite_turns", "Retroactive Rewrite Turns"),
             ("latest_turn_only_rewrite_turns", "Latest-Turn-Only Rewrite Turns"),
         ):
@@ -1665,8 +1714,8 @@ def build_report_markdown(
             "",
             "## Summary",
             "",
-            "| Mode | Raw Tokens | Cache Tokens | Cache Read | Cache Write | Paid Input Tokens | Paid Output Tokens | Paid Input Cost | Cache Read Cost | Cache Write Cost | Paid Output Cost | Total Cost | No-Cache Total Cost | Cache Bust Turns | TTL Expiry Turns | Rewrite Turns | Retroactive Rewrite Turns | Latest-Turn-Only Rewrite Turns | Window Tokens (Cache Counted) | Window Tokens (Cache Reads Excluded) |",
-            "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+            "| Mode | Raw Tokens | Cache Tokens | Cache Read | Cache Write | Paid Input Tokens | Paid Output Tokens | Paid Input Cost | Cache Read Cost | Cache Write Cost | Paid Output Cost | Total Cost | No-Cache Total Cost | Cache Bust Turns | TTL Expiry Turns | Rewrite Turns | Stable Replay Rewrite Turns | Busting Rewrite Turns | Non-Cache-Eligible Rewrite Turns | Retroactive Rewrite Turns | Latest-Turn-Only Rewrite Turns | Window Tokens (Cache Counted) | Window Tokens (Cache Reads Excluded) |",
+            "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
             *rows,
             "",
             "## Impact vs Baseline",
@@ -1712,6 +1761,9 @@ def build_report_html(
             f"<td>{summary.cache_bust_turns:,}</td>"
             f"<td>{summary.ttl_expiry_turns:,}</td>"
             f"<td>{summary.rewrite_turns:,}</td>"
+            f"<td>{summary.stable_replay_rewrite_turns:,}</td>"
+            f"<td>{summary.busting_rewrite_turns:,}</td>"
+            f"<td>{summary.non_cache_eligible_rewrite_turns:,}</td>"
             f"<td>{summary.retroactive_rewrite_turns:,}</td>"
             f"<td>{summary.latest_turn_only_rewrite_turns:,}</td>"
             f"<td>{format_currency(summary.total_cost_usd)}</td>"
@@ -1732,6 +1784,9 @@ def build_report_html(
             ("prompt_window_without_cache_reads", "Window Without Cache Reads"),
             ("cache_bust_turns", "Cache Bust Turns"),
             ("rewrite_turns", "Rewrite Turns"),
+            ("stable_replay_rewrite_turns", "Stable Replay Rewrite Turns"),
+            ("busting_rewrite_turns", "Busting Rewrite Turns"),
+            ("non_cache_eligible_rewrite_turns", "Non-Cache-Eligible Rewrite Turns"),
             ("retroactive_rewrite_turns", "Retroactive Rewrite Turns"),
             ("latest_turn_only_rewrite_turns", "Latest-Turn-Only Rewrite Turns"),
         ):
@@ -1868,7 +1923,7 @@ def build_report_html(
         <table>
           <thead>
             <tr>
-              <th>Mode</th><th>Raw Tokens</th><th>Cache Tokens</th><th>Cache Read</th><th>Cache Write</th><th>Paid Input</th><th>Paid Output</th><th>Cache Busts</th><th>TTL Expiry</th><th>Rewrite Turns</th><th>Retroactive Rewrites</th><th>Latest-Turn-Only Rewrites</th><th>Total Cost</th><th>No-Cache Cost</th><th>Window With Cache</th><th>Window Without Cache Reads</th>
+              <th>Mode</th><th>Raw Tokens</th><th>Cache Tokens</th><th>Cache Read</th><th>Cache Write</th><th>Paid Input</th><th>Paid Output</th><th>Cache Busts</th><th>TTL Expiry</th><th>Rewrite Turns</th><th>Stable Replay Rewrites</th><th>Busting Rewrites</th><th>Non-Cache-Eligible Rewrites</th><th>Retroactive Rewrites</th><th>Latest-Turn-Only Rewrites</th><th>Total Cost</th><th>No-Cache Cost</th><th>Window With Cache</th><th>Window Without Cache Reads</th>
             </tr>
           </thead>
           <tbody>

benchmarks/synthetic_long_cache_suite_report.py

@@ -0,0 +1,428 @@
+#!/usr/bin/env python3
+"""Run a long deterministic synthetic suite for cache and rewrite behavior."""
+
+from __future__ import annotations
+
+import copy
+import html
+import json
+import sys
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from types import SimpleNamespace
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import benchmarks.claude_session_mode_benchmark as bench
+from benchmarks.claude_session_mode_benchmark import (
+    PROXY_MODE_CACHE,
+    PROXY_MODE_TOKEN,
+    ReplayTurn,
+    SessionReplay,
+    determine_winners,
+    format_currency,
+    simulate_replays,
+)
+
+OUTPUT_DIR = Path("benchmark_results") / "synthetic_long_cache_suite"
+MODEL = "claude-sonnet-4-6"
+TTL_MINUTES = 5
+TURNS_PER_SCENARIO = 400
+
+
+class _FakeProvider:
+    @staticmethod
+    def get_context_limit(model: str) -> int:
+        return 200_000
+
+
+class _HistoryPressurePipeline:
+    @staticmethod
+    def apply(messages, **kwargs):  # noqa: ANN001
+        rewritten = []
+        total = len(messages)
+        # Leave the latest two messages untouched; rewrite older tool results.
+        # Token mode reprocesses full history, so prior-turn tool results become
+        # compressed on later turns and can bust prefix cache. Cache mode only
+        # processes the newly-appended delta, so it does not revisit older turns.
+        protected_start = max(total - 2, 0)
+        for index, message in enumerate(messages):
+            content = message.get("content")
+            if (
+                index < protected_start
+                and isinstance(content, list)
+                and any(
+                    isinstance(block, dict) and block.get("type") == "tool_result"
+                    for block in content
+                )
+            ):
+                new_blocks = []
+                for block in content:
+                    if isinstance(block, dict) and block.get("type") == "tool_result":
+                        new_blocks.append({**block, "content": "[compressed-older-tool-result]"})
+                    else:
+                        new_blocks.append(copy.deepcopy(block))
+                rewritten.append({**message, "content": new_blocks})
+            else:
+                rewritten.append(copy.deepcopy(message))
+        return SimpleNamespace(messages=rewritten)
+
+
+class _FakeProxy:
+    def __init__(self) -> None:
+        self.config = SimpleNamespace(image_optimize=False)
+        self.anthropic_provider = _FakeProvider()
+        self.anthropic_pipeline = _HistoryPressurePipeline()
+
+
+def _tool_result_payload(turn_number: int, scenario: str) -> str:
+    return (f"{scenario}-tool-output-{turn_number} " * 80).strip()
+
+
+def _build_stable_append_only() -> SessionReplay:
+    base = datetime(2026, 3, 13, 1, 0, tzinfo=timezone.utc)
+    turns: list[ReplayTurn] = []
+    for index in range(TURNS_PER_SCENARIO):
+        turns.append(
+            ReplayTurn(
+                session_id="stable-append-only",
+                project_key="C--git-synthetic",
+                decoded_project_path=r"C:\git\synthetic",
+                request_id=f"stable-{index + 1:04d}",
+                model=MODEL,
+                timestamp=base + timedelta(minutes=index * 2),
+                input_messages=[
+                    {
+                        "role": "user",
+                        "content": f"Stable append-only turn {index + 1}. Summarize and continue.",
+                    }
+                ],
+                assistant_message={"role": "assistant", "content": f"ok stable {index + 1}"},
+                output_tokens=12,
+            )
+        )
+    return SessionReplay(
+        session_id="stable-append-only",
+        project_key="C--git-synthetic",
+        decoded_project_path=r"C:\git\synthetic",
+        turns=turns,
+    )
+
+
+def _build_token_rewrite_pressure() -> SessionReplay:
+    base = datetime(2026, 3, 14, 1, 0, tzinfo=timezone.utc)
+    turns: list[ReplayTurn] = []
+    for index in range(TURNS_PER_SCENARIO):
+        turn_no = index + 1
+        turns.append(
+            ReplayTurn(
+                session_id="token-rewrite-pressure",
+                project_key="C--git-synthetic",
+                decoded_project_path=r"C:\git\synthetic",
+                request_id=f"rewrite-{turn_no:04d}",
+                model=MODEL,
+                timestamp=base + timedelta(minutes=index * 2),
+                input_messages=[
+                    {"role": "user", "content": f"Inspect tool output for turn {turn_no}."},
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "tool_result",
+                                "tool_use_id": f"tool-{turn_no}",
+                                "content": _tool_result_payload(turn_no, "rewrite"),
+                            }
+                        ],
+                    },
+                ],
+                assistant_message={"role": "assistant", "content": f"ok rewrite {turn_no}"},
+                output_tokens=14,
+            )
+        )
+    return SessionReplay(
+        session_id="token-rewrite-pressure",
+        project_key="C--git-synthetic",
+        decoded_project_path=r"C:\git\synthetic",
+        turns=turns,
+    )
+
+
+def _build_ttl_resets() -> SessionReplay:
+    base = datetime(2026, 3, 15, 1, 0, tzinfo=timezone.utc)
+    turns: list[ReplayTurn] = []
+    for index in range(TURNS_PER_SCENARIO):
+        turns.append(
+            ReplayTurn(
+                session_id="ttl-resets",
+                project_key="C--git-synthetic",
+                decoded_project_path=r"C:\git\synthetic",
+                request_id=f"ttl-{index + 1:04d}",
+                model=MODEL,
+                timestamp=base + timedelta(minutes=index * 7),
+                input_messages=[
+                    {
+                        "role": "user",
+                        "content": f"TTL reset turn {index + 1}. Continue the thread.",
+                    }
+                ],
+                assistant_message={"role": "assistant", "content": f"ok ttl {index + 1}"},
+                output_tokens=12,
+            )
+        )
+    return SessionReplay(
+        session_id="ttl-resets",
+        project_key="C--git-synthetic",
+        decoded_project_path=r"C:\git\synthetic",
+        turns=turns,
+    )
+
+
+def _build_suite() -> list[SessionReplay]:
+    return [
+        _build_stable_append_only(),
+        _build_token_rewrite_pressure(),
+        _build_ttl_resets(),
+    ]
+
+
+def _scenario_label(session_id: str) -> str:
+    return session_id.replace("-", " ").title()
+
+
+def _run_suite() -> tuple[dict[str, dict[str, bench.ModeSummary]], dict[str, bench.ModeSummary]]:
+    original_make_proxy = bench._make_proxy
+    bench._make_proxy = lambda mode: _FakeProxy()
+    try:
+        per_scenario: dict[str, dict[str, bench.ModeSummary]] = {}
+        suite = _build_suite()
+        for replay in suite:
+            _, summaries = simulate_replays([replay], cache_ttl_minutes=TTL_MINUTES)
+            per_scenario[replay.session_id] = summaries
+        _, aggregate = simulate_replays(suite, cache_ttl_minutes=TTL_MINUTES)
+    finally:
+        bench._make_proxy = original_make_proxy
+    return per_scenario, aggregate
+
+
+def _summary_payload(summary: bench.ModeSummary) -> dict[str, int | float | str]:
+    return {
+        "total_cost_usd": summary.total_cost_usd,
+        "no_cache_total_cost_usd": summary.no_cache_total_cost_usd,
+        "forwarded_input_tokens": summary.forwarded_input_tokens,
+        "cache_bust_turns": summary.cache_bust_turns,
+        "ttl_expiry_turns": summary.ttl_expiry_turns,
+        "rewrite_turns": summary.rewrite_turns,
+        "stable_replay_rewrite_turns": summary.stable_replay_rewrite_turns,
+        "busting_rewrite_turns": summary.busting_rewrite_turns,
+        "non_cache_eligible_rewrite_turns": summary.non_cache_eligible_rewrite_turns,
+        "retroactive_rewrite_turns": summary.retroactive_rewrite_turns,
+    }
+
+
+def _write_report(
+    per_scenario: dict[str, dict[str, bench.ModeSummary]],
+    aggregate: dict[str, bench.ModeSummary],
+) -> tuple[Path, Path, Path]:
+    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "turns_per_scenario": TURNS_PER_SCENARIO,
+        "total_turns": TURNS_PER_SCENARIO * len(per_scenario),
+        "ttl_minutes": TTL_MINUTES,
+        "scenarios": {
+            session_id: {
+                mode: _summary_payload(summary)
+                for mode, summary in summaries.items()
+            }
+            for session_id, summaries in per_scenario.items()
+        },
+        "aggregate": {mode: _summary_payload(summary) for mode, summary in aggregate.items()},
+        "aggregate_winners": determine_winners(aggregate),
+    }
+    json_path = OUTPUT_DIR / "synthetic_long_cache_suite.json"
+    md_path = OUTPUT_DIR / "synthetic_long_cache_suite.md"
+    html_path = OUTPUT_DIR / "synthetic_long_cache_suite.html"
+    json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+    md_lines = [
+        "# Synthetic Long Cache Suite",
+        "",
+        f"- Turns per scenario: `{TURNS_PER_SCENARIO}`",
+        f"- Total turns: `{TURNS_PER_SCENARIO * len(per_scenario)}`",
+        f"- Cache TTL: `{TTL_MINUTES}` minutes",
+        "",
+        "## Scenarios",
+        "",
+        "1. `stable-append-only`: append-only conversation, no rewrite pressure",
+        "2. `token-rewrite-pressure`: each turn adds a tool result; older tool results become compressible later",
+        "3. `ttl-resets`: append-only conversation with >TTL gaps to force normal cache expiry",
+        "",
+    ]
+
+    for session_id, summaries in per_scenario.items():
+        winners = determine_winners(summaries)
+        md_lines.extend(
+            [
+                f"## {_scenario_label(session_id)}",
+                "",
+                "| Mode | Cost | Forwarded Tokens | Cache Busts | TTL Expiry | Rewrites | Stable Replay Rewrites | Busting Rewrites | Retroactive Rewrites |",
+                "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+            ]
+        )
+        for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+            summary = summaries[mode]
+            md_lines.append(
+                f"| `{mode}` | {format_currency(summary.total_cost_usd)} | "
+                f"{summary.forwarded_input_tokens:,} | {summary.cache_bust_turns} | "
+                f"{summary.ttl_expiry_turns} | {summary.rewrite_turns} | "
+                f"{summary.stable_replay_rewrite_turns} | {summary.busting_rewrite_turns} | "
+                f"{summary.retroactive_rewrite_turns} |"
+            )
+        md_lines.extend(
+            [
+                "",
+                f"- total cost winner: `{winners['total_cost']}`",
+                f"- no-cache total cost winner: `{winners['no_cache_total_cost']}`",
+                f"- window winner with cache counted: `{winners['window_with_cache']}`",
+                "",
+            ]
+        )
+
+    aggregate_winners = determine_winners(aggregate)
+    md_lines.extend(
+        [
+            "## Aggregate",
+            "",
+            "| Mode | Cost | Forwarded Tokens | Cache Busts | TTL Expiry | Rewrites | Stable Replay Rewrites | Busting Rewrites | Retroactive Rewrites |",
+            "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
+        ]
+    )
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = aggregate[mode]
+        md_lines.append(
+            f"| `{mode}` | {format_currency(summary.total_cost_usd)} | "
+            f"{summary.forwarded_input_tokens:,} | {summary.cache_bust_turns} | "
+            f"{summary.ttl_expiry_turns} | {summary.rewrite_turns} | "
+            f"{summary.stable_replay_rewrite_turns} | {summary.busting_rewrite_turns} | "
+            f"{summary.retroactive_rewrite_turns} |"
+        )
+    md_lines.extend(
+        [
+            "",
+            f"- total cost winner: `{aggregate_winners['total_cost']}`",
+            f"- no-cache total cost winner: `{aggregate_winners['no_cache_total_cost']}`",
+            f"- window winner if cache tokens count: `{aggregate_winners['window_with_cache']}`",
+            f"- window winner if cache read tokens do not count: `{aggregate_winners['window_without_cache_reads']}`",
+        ]
+    )
+    md_path.write_text("\n".join(md_lines), encoding="utf-8")
+
+    scenario_sections: list[str] = []
+    for session_id, summaries in per_scenario.items():
+        rows = []
+        for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+            summary = summaries[mode]
+            rows.append(
+                "<tr>"
+                f"<td><code>{html.escape(mode)}</code></td>"
+                f"<td>{html.escape(format_currency(summary.total_cost_usd))}</td>"
+                f"<td>{summary.forwarded_input_tokens:,}</td>"
+                f"<td>{summary.cache_bust_turns}</td>"
+                f"<td>{summary.ttl_expiry_turns}</td>"
+                f"<td>{summary.rewrite_turns}</td>"
+                f"<td>{summary.stable_replay_rewrite_turns}</td>"
+                f"<td>{summary.busting_rewrite_turns}</td>"
+                f"<td>{summary.retroactive_rewrite_turns}</td>"
+                "</tr>"
+            )
+        scenario_sections.append(
+            "<section class='card'>"
+            f"<h2>{html.escape(_scenario_label(session_id))}</h2>"
+            "<table><thead><tr><th>Mode</th><th>Cost</th><th>Forwarded Tokens</th><th>Cache Busts</th>"
+            "<th>TTL Expiry</th><th>Rewrites</th><th>Stable Replay Rewrites</th>"
+            "<th>Busting Rewrites</th><th>Retroactive Rewrites</th></tr></thead><tbody>"
+            + "".join(rows)
+            + "</tbody></table></section>"
+        )
+
+    aggregate_rows = []
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = aggregate[mode]
+        aggregate_rows.append(
+            "<tr>"
+            f"<td><code>{html.escape(mode)}</code></td>"
+            f"<td>{html.escape(format_currency(summary.total_cost_usd))}</td>"
+            f"<td>{summary.forwarded_input_tokens:,}</td>"
+            f"<td>{summary.cache_bust_turns}</td>"
+            f"<td>{summary.ttl_expiry_turns}</td>"
+            f"<td>{summary.rewrite_turns}</td>"
+            f"<td>{summary.stable_replay_rewrite_turns}</td>"
+            f"<td>{summary.busting_rewrite_turns}</td>"
+            f"<td>{summary.retroactive_rewrite_turns}</td>"
+            "</tr>"
+        )
+
+    html_doc = (
+        "<!doctype html><html><head><meta charset='utf-8'>"
+        "<meta name='viewport' content='width=device-width, initial-scale=1'>"
+        "<title>Synthetic Long Cache Suite</title>"
+        "<style>"
+        "body{font-family:ui-sans-serif,system-ui,sans-serif;max-width:1200px;margin:40px auto;padding:0 20px;line-height:1.55;color:#111827;background:#f8fafc}"
+        "h1,h2{letter-spacing:-0.02em}"
+        "code{background:#e5e7eb;padding:1px 4px;border-radius:4px}"
+        "table{border-collapse:collapse;width:100%;margin:16px 0;background:white}"
+        "th,td{border:1px solid #cbd5e1;padding:10px;text-align:left}"
+        "th{background:#e2e8f0}"
+        ".card{background:white;border:1px solid #cbd5e1;border-radius:16px;padding:24px;margin:18px 0;box-shadow:0 8px 24px rgba(15,23,42,.06)}"
+        "</style></head><body>"
+        "<h1>Synthetic Long Cache Suite</h1>"
+        f"<div class='card'><p>Total turns: <code>{TURNS_PER_SCENARIO * len(per_scenario)}</code><br>"
+        f"Turns per scenario: <code>{TURNS_PER_SCENARIO}</code><br>"
+        f"Cache TTL: <code>{TTL_MINUTES}</code> minutes</p></div>"
+        + "".join(scenario_sections)
+        + "<section class='card'><h2>Aggregate</h2>"
+        "<table><thead><tr><th>Mode</th><th>Cost</th><th>Forwarded Tokens</th><th>Cache Busts</th>"
+        "<th>TTL Expiry</th><th>Rewrites</th><th>Stable Replay Rewrites</th>"
+        "<th>Busting Rewrites</th><th>Retroactive Rewrites</th></tr></thead><tbody>"
+        + "".join(aggregate_rows)
+        + "</tbody></table></section></body></html>"
+    )
+    html_path.write_text(html_doc, encoding="utf-8")
+    return md_path, json_path, html_path
+
+
+def main() -> int:
+    per_scenario, aggregate = _run_suite()
+    md_path, json_path, html_path = _write_report(per_scenario, aggregate)
+    print("Synthetic long cache suite")
+    print(f"turns_per_scenario={TURNS_PER_SCENARIO}")
+    print(f"total_turns={TURNS_PER_SCENARIO * len(per_scenario)}")
+    for session_id, summaries in per_scenario.items():
+        print(f"scenario={session_id}")
+        for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+            summary = summaries[mode]
+            print(
+                f"  {mode}: cost={format_currency(summary.total_cost_usd)} "
+                f"busts={summary.cache_bust_turns} ttl={summary.ttl_expiry_turns} "
+                f"rewrites={summary.rewrite_turns} stable_rw={summary.stable_replay_rewrite_turns} "
+                f"bust_rw={summary.busting_rewrite_turns} "
+                f"forwarded={summary.forwarded_input_tokens}"
+            )
+    print("aggregate")
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = aggregate[mode]
+        print(
+            f"  {mode}: cost={format_currency(summary.total_cost_usd)} "
+            f"busts={summary.cache_bust_turns} ttl={summary.ttl_expiry_turns} "
+            f"rewrites={summary.rewrite_turns} stable_rw={summary.stable_replay_rewrite_turns} "
+            f"bust_rw={summary.busting_rewrite_turns} "
+            f"forwarded={summary.forwarded_input_tokens}"
+        )
+    print(f"Markdown report: {md_path}")
+    print(f"JSON report: {json_path}")
+    print(f"HTML report: {html_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

benchmarks/synthetic_token_cache_bust_report.py

@@ -0,0 +1,370 @@
+#!/usr/bin/env python3
+"""Run a deterministic synthetic replay that forces token-mode cache busts."""
+
+from __future__ import annotations
+
+import copy
+import html
+import json
+import sys
+from datetime import datetime
+from pathlib import Path
+from types import SimpleNamespace
+
+if __package__ in {None, ""}:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import benchmarks.claude_session_mode_benchmark as bench
+from benchmarks.claude_session_mode_benchmark import (
+    PROXY_MODE_CACHE,
+    PROXY_MODE_TOKEN,
+    ReplayTurn,
+    SessionReplay,
+    _apply_mode_to_messages,
+    _cache_gap_within_ttl,
+    determine_winners,
+    format_currency,
+    get_tokenizer,
+    simulate_replays,
+)
+
+OUTPUT_DIR = Path("benchmark_results") / "synthetic_token_cache_bust"
+
+
+class _FakeProvider:
+    @staticmethod
+    def get_context_limit(model: str) -> int:
+        return 200_000
+
+
+class _FakePipeline:
+    @staticmethod
+    def apply(messages, **kwargs):  # noqa: ANN001
+        rewritten = []
+        should_rewrite_history = len(messages) > 2
+        for message in messages:
+            content = message.get("content")
+            if (
+                should_rewrite_history
+                and isinstance(content, list)
+                and any(
+                    isinstance(block, dict) and block.get("type") == "tool_result"
+                    for block in content
+                )
+            ):
+                new_blocks = []
+                for block in content:
+                    if isinstance(block, dict) and block.get("type") == "tool_result":
+                        new_blocks.append({**block, "content": "[compressed-tool-result]"})
+                    else:
+                        new_blocks.append(block)
+                rewritten.append({**message, "content": new_blocks})
+            else:
+                rewritten.append(copy.deepcopy(message))
+        return SimpleNamespace(messages=rewritten)
+
+
+class _FakeProxy:
+    def __init__(self) -> None:
+        self.config = SimpleNamespace(image_optimize=False)
+        self.anthropic_provider = _FakeProvider()
+        self.anthropic_pipeline = _FakePipeline()
+
+
+def _build_replay() -> SessionReplay:
+    return SessionReplay(
+        session_id="token-cache-bust",
+        project_key="C--git-synthetic",
+        decoded_project_path=r"C:\git\synthetic",
+        turns=[
+            ReplayTurn(
+                session_id="token-cache-bust",
+                project_key="C--git-synthetic",
+                decoded_project_path=r"C:\git\synthetic",
+                request_id="r1",
+                model="claude-sonnet-4-6",
+                timestamp=datetime.fromisoformat("2026-03-13T01:00:00+00:00"),
+                input_messages=[
+                    {"role": "user", "content": "Summarize this tool output"},
+                    {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "tool_result",
+                                "tool_use_id": "tool-1",
+                                "content": "X" * 800,
+                            }
+                        ],
+                    },
+                ],
+                assistant_message={"role": "assistant", "content": "ok"},
+                output_tokens=10,
+            ),
+            ReplayTurn(
+                session_id="token-cache-bust",
+                project_key="C--git-synthetic",
+                decoded_project_path=r"C:\git\synthetic",
+                request_id="r2",
+                model="claude-sonnet-4-6",
+                timestamp=datetime.fromisoformat("2026-03-13T01:02:00+00:00"),
+                input_messages=[{"role": "user", "content": "What changed?"}],
+                assistant_message={"role": "assistant", "content": "done"},
+                output_tokens=12,
+            ),
+        ],
+    )
+
+
+def _build_bust_events(replay: SessionReplay) -> dict[str, list[dict[str, object]]]:
+    events: dict[str, list[dict[str, object]]] = {
+        "baseline": [],
+        PROXY_MODE_TOKEN: [],
+        PROXY_MODE_CACHE: [],
+    }
+    ttl_minutes = 5
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        proxy = None if mode == "baseline" else _FakeProxy()
+        prefix_tracker = None if mode == "baseline" else bench.PrefixCacheTracker("anthropic")
+        comp_cache = bench.CompressionCache() if mode == PROXY_MODE_TOKEN else None
+        conversation: list[dict[str, object]] = []
+        previous_original: list[dict[str, object]] | None = None
+        previous_forwarded_context: list[dict[str, object]] | None = None
+        previous_forwarded_request: list[dict[str, object]] | None = None
+        previous_request_id: str | None = None
+        previous_timestamp: datetime | None = None
+
+        for turn in replay.turns:
+            conversation.extend(copy.deepcopy(turn.input_messages))
+            forwarded = _apply_mode_to_messages(
+                proxy,
+                mode,
+                conversation,
+                model=turn.model,
+                prefix_tracker=prefix_tracker,
+                comp_cache=comp_cache,
+                previous_original_messages=previous_original,
+                previous_forwarded_messages=previous_forwarded_context,
+            )
+
+            if (
+                previous_forwarded_request is not None
+                and _cache_gap_within_ttl(
+                    turn.timestamp,
+                    previous_timestamp,
+                    ttl=bench.timedelta(minutes=ttl_minutes),
+                )
+            ):
+                prefix_preserved = (
+                    len(forwarded) >= len(previous_forwarded_request)
+                    and forwarded[: len(previous_forwarded_request)] == previous_forwarded_request
+                )
+                if not prefix_preserved:
+                    divergent_index = next(
+                        (
+                            idx
+                            for idx, (prev_msg, curr_msg) in enumerate(
+                                zip(previous_forwarded_request, forwarded, strict=False)
+                            )
+                            if prev_msg != curr_msg
+                        ),
+                        min(len(previous_forwarded_request), len(forwarded)),
+                    )
+                    events[mode].append(
+                        {
+                            "request_id": turn.request_id,
+                            "previous_request_id": previous_request_id,
+                            "divergent_index": divergent_index,
+                            "previous_forwarded": previous_forwarded_request,
+                            "current_forwarded": forwarded,
+                        }
+                    )
+
+            tokenizer = get_tokenizer(turn.model)
+            if prefix_tracker is not None:
+                bench._update_prefix_tracker(
+                    prefix_tracker,
+                    cache_read_tokens=0,
+                    cache_write_tokens=0,
+                    messages=forwarded,
+                    message_token_counts=[tokenizer.count_message(msg) for msg in forwarded],
+                    original_messages=conversation,
+                )
+
+            conversation.append(copy.deepcopy(turn.assistant_message))
+            previous_original = copy.deepcopy(conversation)
+            previous_forwarded_context = copy.deepcopy(forwarded) + [copy.deepcopy(turn.assistant_message)]
+            previous_forwarded_request = copy.deepcopy(forwarded)
+            previous_request_id = turn.request_id
+            previous_timestamp = turn.timestamp
+
+    return events
+
+
+def _write_report(
+    replay: SessionReplay,
+    summaries: dict[str, bench.ModeSummary],
+    winners: dict[str, str],
+    events: dict[str, list[dict[str, object]]],
+) -> tuple[Path, Path, Path]:
+    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+    payload = {
+        "session_id": replay.session_id,
+        "requests": len(replay.turns),
+        "summaries": {
+            mode: {
+                "total_cost_usd": summary.total_cost_usd,
+                "cache_bust_turns": summary.cache_bust_turns,
+                "rewrite_turns": summary.rewrite_turns,
+                "retroactive_rewrite_turns": summary.retroactive_rewrite_turns,
+                "forwarded_input_tokens": summary.forwarded_input_tokens,
+            }
+            for mode, summary in summaries.items()
+        },
+        "winners": winners,
+        "events": events,
+    }
+    json_path = OUTPUT_DIR / "synthetic_token_cache_bust.json"
+    md_path = OUTPUT_DIR / "synthetic_token_cache_bust.md"
+    html_path = OUTPUT_DIR / "synthetic_token_cache_bust.html"
+    json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+    md_lines = [
+        "# Synthetic Token Cache Bust Report",
+        "",
+        f"Session: `{replay.session_id}`",
+        f"Requests: `{len(replay.turns)}`",
+        "",
+        "## Summary",
+        "",
+        "| Mode | Cost | Cache Busts | Rewrites | Retroactive Rewrites | Forwarded Tokens |",
+        "| --- | ---: | ---: | ---: | ---: | ---: |",
+    ]
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = summaries[mode]
+        md_lines.append(
+            f"| `{mode}` | {format_currency(summary.total_cost_usd)} | "
+            f"{summary.cache_bust_turns} | {summary.rewrite_turns} | "
+            f"{summary.retroactive_rewrite_turns} | {summary.forwarded_input_tokens} |"
+        )
+    md_lines.extend(
+        [
+            "",
+            "## Winners",
+            "",
+            f"- total cost: `{winners['total_cost']}`",
+            f"- no-cache total cost: `{winners['no_cache_total_cost']}`",
+            f"- window with cache counted: `{winners['window_with_cache']}`",
+            f"- window without cache reads: `{winners['window_without_cache_reads']}`",
+            "",
+            "## Cache Bust Events",
+            "",
+        ]
+    )
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        md_lines.append(f"### `{mode}`")
+        if not events[mode]:
+            md_lines.append("")
+            md_lines.append("- none")
+            md_lines.append("")
+            continue
+        md_lines.append("")
+        for event in events[mode]:
+            md_lines.append(
+                f"- request `{event['request_id']}` diverged from `{event['previous_request_id']}` "
+                f"at message index `{event['divergent_index']}`"
+            )
+        md_lines.append("")
+    md_path.write_text("\n".join(md_lines), encoding="utf-8")
+
+    rows = []
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = summaries[mode]
+        rows.append(
+            "<tr>"
+            f"<td>{html.escape(mode)}</td>"
+            f"<td>{html.escape(format_currency(summary.total_cost_usd))}</td>"
+            f"<td>{summary.cache_bust_turns}</td>"
+            f"<td>{summary.rewrite_turns}</td>"
+            f"<td>{summary.retroactive_rewrite_turns}</td>"
+            f"<td>{summary.forwarded_input_tokens}</td>"
+            "</tr>"
+        )
+    event_sections = []
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        section = [f"<h2>{html.escape(mode)}</h2>"]
+        if not events[mode]:
+            section.append("<p>none</p>")
+        else:
+            section.append("<ul>")
+            for event in events[mode]:
+                section.append(
+                    "<li>"
+                    f"request <code>{html.escape(str(event['request_id']))}</code> diverged from "
+                    f"<code>{html.escape(str(event['previous_request_id']))}</code> at message index "
+                    f"<code>{event['divergent_index']}</code>"
+                    "</li>"
+                )
+            section.append("</ul>")
+        event_sections.append("".join(section))
+
+    html_doc = (
+        "<!doctype html><html><head><meta charset='utf-8'>"
+        "<title>Synthetic Token Cache Bust Report</title>"
+        "<style>"
+        "body{font-family:ui-sans-serif,system-ui,sans-serif;margin:32px;line-height:1.5;}"
+        "table{border-collapse:collapse;width:100%;margin:16px 0;}"
+        "th,td{border:1px solid #d0d7de;padding:8px 10px;text-align:left;}"
+        "th{background:#f6f8fa;}"
+        "code{background:#f6f8fa;padding:1px 4px;border-radius:4px;}"
+        "</style></head><body>"
+        "<h1>Synthetic Token Cache Bust Report</h1>"
+        f"<p>Session: <code>{html.escape(replay.session_id)}</code><br>Requests: <code>{len(replay.turns)}</code></p>"
+        "<table><thead><tr><th>Mode</th><th>Cost</th><th>Cache Busts</th><th>Rewrites</th>"
+        "<th>Retroactive Rewrites</th><th>Forwarded Tokens</th></tr></thead><tbody>"
+        + "".join(rows)
+        + "</tbody></table>"
+        "<h2>Winners</h2><ul>"
+        f"<li>total cost: <code>{html.escape(winners['total_cost'])}</code></li>"
+        f"<li>no-cache total cost: <code>{html.escape(winners['no_cache_total_cost'])}</code></li>"
+        f"<li>window with cache counted: <code>{html.escape(winners['window_with_cache'])}</code></li>"
+        f"<li>window without cache reads: <code>{html.escape(winners['window_without_cache_reads'])}</code></li>"
+        "</ul><h2>Cache Bust Events</h2>"
+        + "".join(event_sections)
+        + "</body></html>"
+    )
+    html_path.write_text(html_doc, encoding="utf-8")
+    return md_path, json_path, html_path
+
+
+def main() -> int:
+    original_make_proxy = bench._make_proxy
+    bench._make_proxy = lambda mode: _FakeProxy()
+    try:
+        replay = _build_replay()
+        dataset, summaries = simulate_replays([replay], cache_ttl_minutes=5)
+        events = _build_bust_events(replay)
+    finally:
+        bench._make_proxy = original_make_proxy
+
+    winners = determine_winners(summaries)
+    md_path, json_path, html_path = _write_report(replay, summaries, winners, events)
+    print("Synthetic token-cache-bust replay")
+    print(f"requests={dataset.requests}")
+    for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
+        summary = summaries[mode]
+        print(
+            f"{mode}: cost={format_currency(summary.total_cost_usd)} "
+            f"busts={summary.cache_bust_turns} "
+            f"rewrites={summary.rewrite_turns} "
+            f"retro_rw={summary.retroactive_rewrite_turns} "
+            f"forwarded={summary.forwarded_input_tokens}"
+        )
+    print(f"winner_total_cost={winners['total_cost']}")
+    print(f"Markdown report: {md_path}")
+    print(f"JSON report: {json_path}")
+    print(f"HTML report: {html_path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

docs/benchmarks.md

@@ -228,6 +228,50 @@ This benchmark compares `token` vs `cache` proxy modes on the same synthetic con
 
 Use it when you want a clean PR-vs-`main` comparison on the same transcript slice.
 
+For a deterministic cache-busting proof case, run:
+
+```bash
+python benchmarks/synthetic_token_cache_bust_report.py
+```
+
+That synthetic replay forces `token` mode to retroactively rewrite a prior tool result on the second turn while `cache` mode remains stable. Use it to verify the simulator can distinguish:
+
+- `token`: history rewrite + cache bust
+- `cache`: no rewrite + no bust
+
+For a reproducible local report bundle that combines:
+
+- full real-session replay summaries
+- local-only processed real input/output excerpts
+- synthetic token-bust proof
+- synthetic long-form stress tests
+
+run:
+
+```bash
+python benchmarks/cache_validation_bundle.py --workers 1 --output-dir benchmark_results/cache_validation_bundle_full
+```
+
+Notes:
+
+- By default the bundle is redaction-safe for sharing:
+  - real processed reports redact transcript-derived content excerpts
+  - manifest paths are redacted
+- To include local processed content excerpts for private review on your own machine:
+
+```bash
+python benchmarks/cache_validation_bundle.py --workers 1 --include-content
+```
+
+- The bundle writes:
+  - `index.html` / `index.md`: top-level summary and links
+  - `bundle_manifest.json`: runtime metadata + corpus fingerprint
+  - `real/`: full real-session replay reports
+  - `real_processed/`: processed before/after excerpts from real transcripts
+  - `synthetic_token_bust/`: minimal explicit cache-bust proof
+  - `synthetic_long_suite/`: long deterministic rewrite/TTL scenarios
+- Checkpoints are scoped under the bundle output directory and fingerprinted by the selected corpus so stale runs do not contaminate new results.
+
 The Claude session benchmark replays local transcript data from `~/.claude/projects`
 through `baseline`, `token`, and `cache` modes. It estimates raw tokens, cache
 read/write tokens, paid input/output costs, and prompt-window winners under two

headroom/providers/anthropic.py

@@ -250,7 +250,13 @@ class AnthropicTokenCounter(TokenCounter):
 
         if self._encoding:
             # tiktoken with ~1.1x multiplier for Claude
-            base_count = len(self._encoding.encode(text))
+            try:
+                base_count = len(self._encoding.encode(text))
+            except ValueError:
+                # Real tool output can legitimately contain strings that look like
+                # tiktoken special tokens (for example FIM markers in code spans).
+                # Treat them as ordinary text for estimation instead of failing.
+                base_count = len(self._encoding.encode(text, disallowed_special=()))
             return int(base_count * 1.1)
 
         # Character-based fallback

tests/test_claude_session_mode_benchmark.py

@@ -338,9 +338,9 @@ def test_determine_winners_includes_no_cache_counterfactual() -> None:
 def test_resolve_checkpoint_dir_namespaces_sampling_mode() -> None:
     base = Path("benchmark_results") / "checkpoints"
 
-    assert resolve_checkpoint_dir(base).name == "v4__ttl_5m__full"
+    assert resolve_checkpoint_dir(base).name == "v5__ttl_5m__full"
     assert (
-        resolve_checkpoint_dir(base, recent_turns_per_session=200).name == "v4__ttl_5m__recent_200"
+        resolve_checkpoint_dir(base, recent_turns_per_session=200).name == "v5__ttl_5m__recent_200"
     )
 
 
@@ -519,6 +519,11 @@ def test_synthetic_token_mode_busts_cache_while_cache_mode_stays_stable(monkeypa
 
     assert token.cache_bust_turns == 1
     assert token.rewrite_turns >= 1
+    assert token.busting_rewrite_turns >= 1
+    assert token.non_cache_eligible_rewrite_turns == 0
+    assert token.stable_replay_rewrite_turns == 0
     assert token.retroactive_rewrite_turns >= 1
     assert cache.cache_bust_turns == 0
+    assert cache.busting_rewrite_turns == 0
+    assert cache.non_cache_eligible_rewrite_turns == 0
     assert cache.retroactive_rewrite_turns == 0

tests/test_providers/test_anthropic.py

@@ -22,6 +22,11 @@ class TestAnthropicTokenCounting:
         count = counter.count_messages(messages)
         assert count > 0
 
+    def test_count_text_allows_literal_special_tokens(self, anthropic_provider):
+        counter = anthropic_provider.get_token_counter("claude-3-5-sonnet-20241022")
+        count = counter.count_text("prefix <|fim_suffix|> suffix")
+        assert count > 0
+
 
 class TestAnthropicModelLimits:
     @pytest.fixture