akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Harden anthropic cache-mode replay stability

Browse source
This commit is contained in:
JerrettDavis 2026-04-05 16:11:39 -05:00
parent d00c6739e1
commit 5ffa77f4a3
7 changed files with 1518 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

595
benchmarks/claude_session_branch_compare.py Normal file
View file

@ -0,0 +1,595 @@
#!/usr/bin/env python3
"""Compare Claude session mode simulations across two git refs."""
from __future__ import annotations
import argparse
import json
import os
import shutil
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]))
from benchmarks.claude_session_mode_benchmark import (
IMPACT_DIRECTION,
OUTPUT_JSON,
PROXY_MODE_CACHE,
PROXY_MODE_TOKEN,
format_currency,
)
DEFAULT_OUTPUT_DIR = Path("benchmark_results") / "branch_compare"
@dataclass
class BranchResult:
ref: str
label: str
commit: str
summary: str
dataset: dict[str, Any]
observed: dict[str, Any]
summaries: dict[str, dict[str, Any]]
winners: dict[str, str]
output_dir: str
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--left-ref", default="upstream/main")
parser.add_argument("--right-ref", default="HEAD")
parser.add_argument("--left-label", default="main")
parser.add_argument("--right-label", default="pr")
parser.add_argument("--root", type=Path, default=Path.home() / ".claude" / "projects")
parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
parser.add_argument("--max-sessions", type=int, default=None)
parser.add_argument("--recent-turns-per-session", type=int, default=None)
parser.add_argument("--cache-ttl-minutes", type=int, default=5)
parser.add_argument("--cache-write-multiplier", type=float, default=1.25)
parser.add_argument("--workers", type=int, default=1)
parser.add_argument(
"--python",
default=sys.executable,
help="Python executable to use inside each worktree.",
)
parser.add_argument(
"--keep-worktrees",
action="store_true",
help="Do not remove temporary worktrees after the comparison run.",
)
return parser.parse_args()
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 _branch_output_dir(base: Path, label: str) -> Path:
return base / _ref_slug(label)
def _comparison_paths(base: Path) -> tuple[Path, Path, Path]:
return (
base / "claude_session_branch_compare.md",
base / "claude_session_branch_compare.json",
base / "claude_session_branch_compare.html",
)
def _mode_metric(branch: BranchResult, mode: str, field: str) -> float:
summary = branch.summaries[mode]
if field == "no_cache_total_cost_usd":
if "no_cache_total_cost_usd" in summary:
value = summary["no_cache_total_cost_usd"]
else:
value = (
float(summary["paid_input_cost_usd"])
+ (float(summary["cache_read_cost_usd"]) * 10.0)
+ float(summary["paid_output_cost_usd"])
)
elif field == "prompt_window_with_cache":
value = float(summary["forwarded_input_tokens"])
elif field == "prompt_window_without_cache_reads":
value = float(summary["forwarded_input_tokens"]) - float(summary["cache_read_tokens"])
else:
value = summary[field]
if isinstance(value, bool):
return float(value)
return float(value)
def _delta(left: float, right: float) -> float:
return right - left
def _classify_delta(field: str, delta: float) -> str:
direction = IMPACT_DIRECTION.get(field, "same")
tolerance = 1e-9
if abs(delta) <= tolerance:
return "no_change"
if direction == "lower":
return "assist" if delta < 0 else "harm"
if direction == "higher":
return "assist" if delta > 0 else "harm"
return "harm"
def _build_benchmark_command(
python_executable: str,
script_path: Path,
root: Path,
output_dir: Path,
max_sessions: int | None,
recent_turns_per_session: int | None,
cache_ttl_minutes: int,
cache_write_multiplier: float,
workers: int,
) -> list[str]:
command = [
python_executable,
str(script_path),
"--root",
str(root),
"--output-dir",
str(output_dir),
"--cache-ttl-minutes",
str(cache_ttl_minutes),
"--cache-write-multiplier",
str(cache_write_multiplier),
"--workers",
str(workers),
]
if max_sessions is not None:
command.extend(["--max-sessions", str(max_sessions)])
if recent_turns_per_session is not None:
command.extend(["--recent-turns-per-session", str(recent_turns_per_session)])
return command
def _load_branch_result(
repo_root: Path,
ref: str,
label: str,
branch_output_dir: Path,
) -> BranchResult:
payload = json.loads((branch_output_dir / OUTPUT_JSON).read_text(encoding="utf-8"))
commit = _run_git(["rev-parse", ref], repo_root)
summary = _run_git(["show", "-s", "--format=%s", ref], repo_root)
return BranchResult(
ref=ref,
label=label,
commit=commit,
summary=summary,
dataset=payload["dataset"],
observed=payload["observed"],
summaries=payload["summaries"],
winners=payload["winners"],
output_dir=str(branch_output_dir),
)
def _run_branch_benchmark(
repo_root: Path,
ref: str,
label: str,
args: argparse.Namespace,
worktree_root: Path,
) -> BranchResult:
worktree_dir = worktree_root / _ref_slug(label)
branch_output_dir = _branch_output_dir(args.output_dir, label)
branch_output_dir.mkdir(parents=True, exist_ok=True)
if worktree_dir.exists():
shutil.rmtree(worktree_dir)
_run_git(["worktree", "add", "--detach", str(worktree_dir), ref], repo_root)
try:
command = _build_benchmark_command(
python_executable=args.python,
script_path=repo_root / "benchmarks" / "claude_session_mode_benchmark.py",
root=args.root,
output_dir=branch_output_dir,
max_sessions=args.max_sessions,
recent_turns_per_session=args.recent_turns_per_session,
cache_ttl_minutes=args.cache_ttl_minutes,
cache_write_multiplier=args.cache_write_multiplier,
workers=args.workers,
)
env = os.environ.copy()
env["PYTHONPATH"] = os.pathsep.join(
[str(worktree_dir), str(repo_root), env.get("PYTHONPATH", "")]
).rstrip(os.pathsep)
subprocess.run(command, cwd=worktree_dir, check=True, env=env)
return _load_branch_result(repo_root, ref, label, branch_output_dir)
finally:
if not args.keep_worktrees:
subprocess.run(
["git", "worktree", "remove", "--force", str(worktree_dir)],
cwd=repo_root,
check=True,
)
def _winner_line(metric: str, left: BranchResult, right: BranchResult) -> str:
left_winner = left.winners[metric]
right_winner = right.winners[metric]
if left_winner == right_winner:
return f"- {metric}: both pick `{left_winner}`"
return (
f"- {metric}: `{left.label}` picks `{left_winner}`, `{right.label}` picks `{right_winner}`"
)
def _build_six_way_rows(
left: BranchResult, right: BranchResult
) -> list[dict[str, str | float | int]]:
rows: list[dict[str, str | float | int]] = []
for branch in (left, right):
for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
summary = branch.summaries[mode]
cost_delta = _mode_metric(branch, mode, "total_cost_usd") - _mode_metric(
branch, "baseline", "total_cost_usd"
)
window_delta = int(
_mode_metric(branch, mode, "prompt_window_with_cache")
- _mode_metric(branch, "baseline", "prompt_window_with_cache")
)
read_delta = int(
_mode_metric(branch, mode, "cache_read_tokens")
- _mode_metric(branch, "baseline", "cache_read_tokens")
)
write_delta = int(
_mode_metric(branch, mode, "cache_write_tokens")
- _mode_metric(branch, "baseline", "cache_write_tokens")
)
paid_input_delta = int(
_mode_metric(branch, mode, "regular_input_tokens")
- _mode_metric(branch, "baseline", "regular_input_tokens")
)
rows.append(
{
"branch": branch.label,
"mode": mode,
"forwarded_input_tokens": int(summary["forwarded_input_tokens"]),
"cache_read_tokens": int(summary["cache_read_tokens"]),
"cache_write_tokens": int(summary["cache_write_tokens"]),
"regular_input_tokens": int(summary["regular_input_tokens"]),
"output_tokens": int(summary["output_tokens"]),
"total_cost_usd": float(summary["total_cost_usd"]),
"cost_delta_vs_branch_baseline": cost_delta,
"window_delta_vs_branch_baseline": window_delta,
"cache_read_delta_vs_branch_baseline": read_delta,
"cache_write_delta_vs_branch_baseline": write_delta,
"paid_input_delta_vs_branch_baseline": paid_input_delta,
"is_branch_winner": "yes" if branch.winners["total_cost"] == mode else "no",
}
)
return rows
def build_compare_markdown(left: BranchResult, right: BranchResult) -> str:
six_way_rows = _build_six_way_rows(left, right)
lines = [
"# Claude Session Branch Comparison",
"",
"## Branches",
"",
f"- {left.label}: `{left.ref}` @ `{left.commit[:12]}` - {left.summary}",
f"- {right.label}: `{right.ref}` @ `{right.commit[:12]}` - {right.summary}",
"",
"## Dataset",
"",
f"- Projects: {right.dataset['projects']}",
f"- Sessions: {right.dataset['sessions']}",
f"- Requests: {right.dataset['requests']}",
f"- Sampled requests: {right.dataset.get('sampled_requests', 0)}",
f"- Sampling: {right.dataset.get('sampling_note', 'Full sessions')}",
"",
"## Winner Comparison",
"",
_winner_line("total_cost", left, right),
_winner_line("no_cache_total_cost", left, right),
_winner_line("window_with_cache", left, right),
_winner_line("window_without_cache_reads", left, right),
"",
"## Six-Way Mode Matrix",
"",
"| Branch | Mode | Forwarded Input | Cache Read | Cache Write | Paid Input | Paid Output | Total Cost | Cost Δ vs Branch Baseline | Window Δ vs Branch Baseline | Winner |",
"| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- |",
*[
"| "
+ " | ".join(
[
str(row["branch"]),
str(row["mode"]),
f"{int(row['forwarded_input_tokens']):,}",
f"{int(row['cache_read_tokens']):,}",
f"{int(row['cache_write_tokens']):,}",
f"{int(row['regular_input_tokens']):,}",
f"{int(row['output_tokens']):,}",
format_currency(float(row["total_cost_usd"])),
format_currency(float(row["cost_delta_vs_branch_baseline"])),
f"{int(row['window_delta_vs_branch_baseline']):,}",
str(row["is_branch_winner"]),
]
)
+ " |"
for row in six_way_rows
],
"",
"## Mode Deltas",
"",
f"| Mode | Metric | {left.label} | {right.label} | Delta ({right.label} - {left.label}) | Classification |",
"| --- | --- | ---: | ---: | ---: | --- |",
]
metrics = [
("total_cost_usd", "Total Cost", format_currency),
("no_cache_total_cost_usd", "No-Cache Total Cost", format_currency),
("forwarded_input_tokens", "Forwarded Input Tokens", lambda v: f"{int(v):,}"),
("cache_read_tokens", "Cache Read Tokens", lambda v: f"{int(v):,}"),
("cache_write_tokens", "Cache Write Tokens", lambda v: f"{int(v):,}"),
("cache_bust_turns", "Cache Bust Turns", lambda v: f"{int(v):,}"),
("ttl_expiry_turns", "TTL Expiry Turns", lambda v: f"{int(v):,}"),
("prompt_window_with_cache", "Window With Cache", lambda v: f"{int(v):,}"),
(
"prompt_window_without_cache_reads",
"Window Without Cache Reads",
lambda v: f"{int(v):,}",
),
]
for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
for field, label, formatter in metrics:
left_value = _mode_metric(left, mode, field)
right_value = _mode_metric(right, mode, field)
delta = _delta(left_value, right_value)
delta_text = format_currency(delta) if "cost" in field else f"{int(delta):,}"
classification = _classify_delta(field, delta)
lines.append(
f"| {mode} | {label} | {formatter(left_value)} | {formatter(right_value)} | {delta_text} | {classification} |"
)
return "\n".join(lines)
def build_compare_html(left: BranchResult, right: BranchResult) -> str:
six_way_rows = []
for row in _build_six_way_rows(left, right):
six_way_rows.append(
"<tr>"
f"<td>{row['branch']}</td>"
f"<td><span class='pill'>{row['mode']}</span></td>"
f"<td>{int(row['forwarded_input_tokens']):,}</td>"
f"<td>{int(row['cache_read_tokens']):,}</td>"
f"<td>{int(row['cache_write_tokens']):,}</td>"
f"<td>{int(row['regular_input_tokens']):,}</td>"
f"<td>{int(row['output_tokens']):,}</td>"
f"<td>{format_currency(float(row['total_cost_usd']))}</td>"
f"<td>{format_currency(float(row['cost_delta_vs_branch_baseline']))}</td>"
f"<td>{int(row['window_delta_vs_branch_baseline']):,}</td>"
f"<td>{row['is_branch_winner']}</td>"
"</tr>"
)
cards = []
for branch in (left, right):
cards.append(
"<div class='card'>"
f"<div class='eyebrow'>{branch.label}</div>"
f"<h2>{branch.ref}</h2>"
f"<p><code>{branch.commit[:12]}</code></p>"
f"<p>{branch.summary}</p>"
"<div class='winner-grid'>"
f"<div><span>Total Cost</span><strong>{branch.winners['total_cost']}</strong></div>"
f"<div><span>No Cache</span><strong>{branch.winners['no_cache_total_cost']}</strong></div>"
f"<div><span>Window + Cache</span><strong>{branch.winners['window_with_cache']}</strong></div>"
"<div><span>Window - Reads</span>"
f"<strong>{branch.winners['window_without_cache_reads']}</strong></div>"
"</div>"
"</div>"
)
rows = []
for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
for field, label in (
("total_cost_usd", "Total Cost"),
("no_cache_total_cost_usd", "No-Cache Total Cost"),
("forwarded_input_tokens", "Forwarded Input Tokens"),
("cache_read_tokens", "Cache Read Tokens"),
("cache_write_tokens", "Cache Write Tokens"),
("cache_bust_turns", "Cache Bust Turns"),
("prompt_window_with_cache", "Window With Cache"),
("prompt_window_without_cache_reads", "Window Without Cache Reads"),
):
left_value = _mode_metric(left, mode, field)
right_value = _mode_metric(right, mode, field)
delta = _delta(left_value, right_value)
is_cost = "cost" in field
formatter = format_currency if is_cost else (lambda v: f"{int(v):,}")
delta_text = format_currency(delta) if is_cost else f"{int(delta):,}"
delta_class = "pos" if delta > 0 else "neg" if delta < 0 else "neutral"
classification = _classify_delta(field, delta)
rows.append(
"<tr>"
f"<td><span class='pill'>{mode}</span></td>"
f"<td>{label}</td>"
f"<td>{formatter(left_value)}</td>"
f"<td>{formatter(right_value)}</td>"
f"<td class='{delta_class}'>{delta_text}</td>"
f"<td>{classification}</td>"
"</tr>"
)
return f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Claude Session Branch Comparison</title>
<style>
:root {{
--bg: #f8fafc;
--fg: #020617;
--muted: #475569;
--card: #ffffff;
--line: #e2e8f0;
--soft: #f1f5f9;
--accent: #0f172a;
--accent-soft: #e2e8f0;
--good: #166534;
--bad: #991b1b;
--shadow: 0 10px 30px rgba(15, 23, 42, 0.08);
--radius: 16px;
--font: "Geist", "Segoe UI", system-ui, sans-serif;
}}
* {{ box-sizing: border-box; }}
body {{ margin: 0; background: var(--bg); color: var(--fg); font-family: var(--font); }}
.shell {{ max-width: 1280px; margin: 0 auto; padding: 32px 16px 56px; }}
.hero, .card {{
background: var(--card);
border: 1px solid var(--line);
border-radius: var(--radius);
box-shadow: var(--shadow);
}}
.hero {{ padding: 24px; }}
.eyebrow {{ color: var(--muted); font-size: 12px; font-weight: 600; text-transform: uppercase; letter-spacing: .08em; }}
h1, h2 {{ margin: 0; letter-spacing: -0.03em; }}
p {{ color: var(--muted); line-height: 1.5; }}
.grid {{ display: grid; gap: 16px; margin-top: 16px; }}
.two {{ grid-template-columns: repeat(auto-fit, minmax(320px, 1fr)); }}
.card {{ padding: 20px; }}
.winner-grid {{ display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 12px; margin-top: 16px; }}
.winner-grid span {{ display: block; color: var(--muted); font-size: 12px; }}
.winner-grid strong {{ display: block; margin-top: 4px; font-size: 16px; }}
.table-card {{ margin-top: 16px; padding: 0; overflow: hidden; }}
.table-wrap {{ overflow-x: auto; }}
table {{ width: 100%; border-collapse: collapse; }}
th, td {{ padding: 12px 14px; border-bottom: 1px solid var(--line); text-align: left; white-space: nowrap; }}
th {{ background: var(--soft); color: var(--muted); font-size: 12px; text-transform: uppercase; letter-spacing: .04em; }}
.pill {{
display: inline-flex; align-items: center; border-radius: 999px; padding: 4px 10px;
background: var(--accent-soft); color: var(--accent); font-size: 12px; font-weight: 600;
}}
.pos {{ color: var(--bad); font-weight: 600; }}
.neg {{ color: var(--good); font-weight: 600; }}
.neutral {{ color: var(--muted); }}
code {{ font-family: ui-monospace, SFMono-Regular, Consolas, monospace; }}
</style>
</head>
<body>
<div class="shell">
<section class="hero">
<div class="eyebrow">Branch Comparison</div>
<h1>Claude Session Mode Simulation</h1>
<p>Same local Claude transcript corpus. Same simulation knobs. Two git refs. This report isolates code-level behavior changes between the branches.</p>
<div class="grid two">
{"".join(cards)}
</div>
</section>
<section class="card table-card">
<div class="table-wrap">
<table>
<thead>
<tr>
<th>Branch</th>
<th>Mode</th>
<th>Forwarded Input</th>
<th>Cache Read</th>
<th>Cache Write</th>
<th>Paid Input</th>
<th>Paid Output</th>
<th>Total Cost</th>
<th>Cost Δ vs Branch Baseline</th>
<th>Window Δ vs Branch Baseline</th>
<th>Winner</th>
</tr>
</thead>
<tbody>
{"".join(six_way_rows)}
</tbody>
</table>
</div>
</section>
<section class="card table-card">
<div class="table-wrap">
<table>
<thead>
<tr>
<th>Mode</th>
<th>Metric</th>
<th>{left.label}</th>
<th>{right.label}</th>
<th>Delta</th>
<th>Classification</th>
</tr>
</thead>
<tbody>
{"".join(rows)}
</tbody>
</table>
</div>
</section>
</div>
</body>
</html>"""
def write_compare_report(
output_dir: Path,
left: BranchResult,
right: BranchResult,
) -> tuple[Path, Path, Path]:
output_dir.mkdir(parents=True, exist_ok=True)
md_path, json_path, html_path = _comparison_paths(output_dir)
md_path.write_text(build_compare_markdown(left, right), encoding="utf-8")
html_path.write_text(build_compare_html(left, right), encoding="utf-8")
payload = {
"left": asdict(left),
"right": asdict(right),
"left_winners": left.winners,
"right_winners": right.winners,
}
json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
return md_path, json_path, html_path
def main() -> int:
args = parse_args()
repo_root = Path(__file__).resolve().parents[1]
if not args.output_dir.is_absolute():
args.output_dir = (repo_root / args.output_dir).resolve()
if not args.root.is_absolute():
args.root = args.root.resolve()
args.output_dir.mkdir(parents=True, exist_ok=True)
worktree_root = Path(tempfile.mkdtemp(prefix="headroom-branch-compare-"))
try:
left = _run_branch_benchmark(repo_root, args.left_ref, args.left_label, args, worktree_root)
right = _run_branch_benchmark(
repo_root, args.right_ref, args.right_label, args, worktree_root
)
md_path, json_path, html_path = write_compare_report(args.output_dir, left, right)
print(f"Compared {left.label} ({left.ref}) vs {right.label} ({right.ref})")
print(f"Markdown report: {md_path}")
print(f"JSON report: {json_path}")
print(f"HTML report: {html_path}")
return 0
finally:
if args.keep_worktrees:
print(f"Retained worktrees under {worktree_root}")
else:
shutil.rmtree(worktree_root, ignore_errors=True)
if __name__ == "__main__":
raise SystemExit(main())

409
benchmarks/claude_session_mode_benchmark.py
View file

@ -20,11 +20,16 @@ from headroom.cache.prefix_tracker import PrefixCacheTracker
from headroom.pricing.litellm_pricing import get_model_pricing
from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin
from headroom.proxy.models import ProxyConfig
from headroom.proxy.modes import PROXY_MODE_CACHE, PROXY_MODE_TOKEN
from headroom.proxy.server import HeadroomProxy
from headroom.tokenizers import get_tokenizer
from headroom.utils import extract_user_query
try:
from headroom.proxy.modes import PROXY_MODE_CACHE, PROXY_MODE_TOKEN
except ImportError:
PROXY_MODE_CACHE = "cache"
PROXY_MODE_TOKEN = "token"
DEFAULT_ROOT = Path.home() / ".claude" / "projects"
DEFAULT_OUTPUT_DIR = Path("benchmark_results")
DEFAULT_CACHE_TTL_MINUTES = 5
@ -96,6 +101,9 @@ class ModeSummary:
cache_eligible_turns: int = 0
cache_bust_turns: int = 0
ttl_expiry_turns: int = 0
rewrite_turns: int = 0
retroactive_rewrite_turns: int = 0
latest_turn_only_rewrite_turns: int = 0
turns: list[TurnMetrics] = field(default_factory=list)
@property
@ -136,6 +144,24 @@ class DatasetSummary:
sampling_note: str = ""
IMPACT_DIRECTION = {
"forwarded_input_tokens": "lower",
"cache_read_tokens": "higher",
"cache_write_tokens": "lower",
"regular_input_tokens": "lower",
"output_tokens": "same",
"total_cost_usd": "lower",
"no_cache_total_cost_usd": "lower",
"prompt_window_with_cache": "lower",
"prompt_window_without_cache_reads": "lower",
"cache_bust_turns": "lower",
"ttl_expiry_turns": "lower",
"rewrite_turns": "lower",
"retroactive_rewrite_turns": "lower",
"latest_turn_only_rewrite_turns": "lower",
}
@dataclass
class ObservedSummary:
sessions: int = 0
@ -221,6 +247,9 @@ def _mode_summary_from_dict(data: dict[str, Any]) -> ModeSummary:
cache_eligible_turns=data.get("cache_eligible_turns", 0),
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),
retroactive_rewrite_turns=data.get("retroactive_rewrite_turns", 0),
latest_turn_only_rewrite_turns=data.get("latest_turn_only_rewrite_turns", 0),
turns=turns,
)
return summary
@ -618,6 +647,133 @@ def _common_prefix_tokens(
return common
def _rewrite_scope(
original_messages: list[dict[str, Any]],
forwarded_messages: list[dict[str, Any]],
*,
stable_prefix_message_count: int,
) -> tuple[bool, bool]:
if original_messages == forwarded_messages:
return False, False
stable_count = min(
stable_prefix_message_count,
len(original_messages),
len(forwarded_messages),
)
retroactive = False
if len(forwarded_messages) < stable_prefix_message_count:
retroactive = True
elif stable_count > 0 and forwarded_messages[:stable_count] != original_messages[:stable_count]:
retroactive = True
return True, retroactive
def _extract_cache_stable_delta(
current_messages: list[dict[str, Any]],
previous_original_messages: list[dict[str, Any]] | None,
previous_forwarded_messages: list[dict[str, Any]] | None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]] | None:
if previous_original_messages is None or previous_forwarded_messages is None:
return None
if len(current_messages) < len(previous_original_messages):
return None
stable_count = len(previous_original_messages)
if current_messages[:stable_count] != previous_original_messages:
return None
return (
copy.deepcopy(previous_forwarded_messages),
copy.deepcopy(current_messages[stable_count:]),
)
def _extract_cache_stable_last_message_suffix(
current_messages: list[dict[str, Any]],
previous_original_messages: list[dict[str, Any]] | None,
previous_forwarded_messages: list[dict[str, Any]] | None,
) -> tuple[list[dict[str, Any]], dict[str, Any], list[dict[str, Any]]] | None:
if not previous_original_messages or previous_forwarded_messages is None:
return None
if (
len(current_messages) != len(previous_original_messages)
or len(previous_forwarded_messages) != len(previous_original_messages)
or not current_messages
):
return None
prefix_len = len(current_messages) - 1
if prefix_len > 0 and current_messages[:prefix_len] != previous_original_messages[:prefix_len]:
return None
current_last = current_messages[-1]
previous_original_last = previous_original_messages[-1]
previous_forwarded_last = previous_forwarded_messages[-1]
if current_last.get("role") != previous_original_last.get("role") or current_last.get(
"role"
) != previous_forwarded_last.get("role"):
return None
current_content = current_last.get("content")
previous_original_content = previous_original_last.get("content")
previous_forwarded_content = previous_forwarded_last.get("content")
if (
isinstance(current_content, str)
and isinstance(previous_original_content, str)
and isinstance(previous_forwarded_content, str)
and current_content.startswith(previous_original_content)
):
suffix = current_content[len(previous_original_content) :]
delta_messages = []
if suffix:
delta_messages = [{**copy.deepcopy(current_last), "content": suffix}]
return (
copy.deepcopy(previous_forwarded_messages[:-1]),
copy.deepcopy(previous_forwarded_last),
delta_messages,
)
if (
isinstance(current_content, list)
and isinstance(previous_original_content, list)
and isinstance(previous_forwarded_content, list)
and len(current_content) >= len(previous_original_content)
and current_content[: len(previous_original_content)] == previous_original_content
):
delta_blocks = copy.deepcopy(current_content[len(previous_original_content) :])
delta_messages = []
if delta_blocks:
delta_messages = [{**copy.deepcopy(current_last), "content": delta_blocks}]
return (
copy.deepcopy(previous_forwarded_messages[:-1]),
copy.deepcopy(previous_forwarded_last),
delta_messages,
)
return None
def _merge_appended_message_delta(
previous_forwarded_message: dict[str, Any],
delta_forwarded_message: dict[str, Any] | None,
) -> dict[str, Any] | None:
if delta_forwarded_message is None:
return copy.deepcopy(previous_forwarded_message)
if previous_forwarded_message.get("role") != delta_forwarded_message.get("role"):
return None
previous_content = previous_forwarded_message.get("content")
delta_content = delta_forwarded_message.get("content")
if isinstance(previous_content, str) and isinstance(delta_content, str):
return {
**copy.deepcopy(previous_forwarded_message),
"content": previous_content + delta_content,
}
if isinstance(previous_content, list) and isinstance(delta_content, list):
return {
**copy.deepcopy(previous_forwarded_message),
"content": copy.deepcopy(previous_content) + copy.deepcopy(delta_content),
}
return None
def _make_proxy(mode: str) -> HeadroomProxy:
cfg = ProxyConfig(
mode=mode,
@ -655,25 +811,47 @@ def _apply_mode_to_messages(
assert proxy is not None
assert prefix_tracker is not None
if mode == PROXY_MODE_CACHE:
delta = AnthropicHandlerMixin._extract_cache_stable_delta(
supports_delta_replay = hasattr(
AnthropicHandlerMixin, "_extract_cache_stable_last_message_suffix"
)
if not supports_delta_replay:
frozen_message_count = prefix_tracker.get_frozen_message_count()
context_limit = proxy.anthropic_provider.get_context_limit(model)
result = proxy.anthropic_pipeline.apply(
messages=copy.deepcopy(messages),
model=model,
model_limit=context_limit,
context=extract_user_query(messages),
frozen_message_count=frozen_message_count,
)
if hasattr(AnthropicHandlerMixin, "_restore_frozen_prefix"):
result.messages, _ = AnthropicHandlerMixin._restore_frozen_prefix(
messages,
result.messages,
frozen_message_count=frozen_message_count,
)
return result.messages
delta = _extract_cache_stable_delta(
messages,
previous_original_messages,
previous_forwarded_messages,
)
if delta is None:
return copy.deepcopy(messages)
stable_forwarded_prefix, delta_messages = delta
if not delta_messages:
return stable_forwarded_prefix
context_limit = proxy.anthropic_provider.get_context_limit(model)
result = proxy.anthropic_pipeline.apply(
messages=delta_messages,
model=model,
model_limit=context_limit,
context=extract_user_query(delta_messages),
frozen_message_count=0,
)
return stable_forwarded_prefix + result.messages
if delta is not None:
stable_forwarded_prefix, delta_messages = delta
if not delta_messages:
return stable_forwarded_prefix
context_limit = proxy.anthropic_provider.get_context_limit(model)
result = proxy.anthropic_pipeline.apply(
messages=delta_messages,
model=model,
model_limit=context_limit,
context=extract_user_query(delta_messages),
frozen_message_count=0,
)
return stable_forwarded_prefix + result.messages
return copy.deepcopy(messages)
frozen_message_count = prefix_tracker.get_frozen_message_count()
@ -842,6 +1020,9 @@ def _merge_mode_summary(target: ModeSummary, source: ModeSummary) -> None:
target.cache_eligible_turns += source.cache_eligible_turns
target.cache_bust_turns += source.cache_bust_turns
target.ttl_expiry_turns += source.ttl_expiry_turns
target.rewrite_turns += source.rewrite_turns
target.retroactive_rewrite_turns += source.retroactive_rewrite_turns
target.latest_turn_only_rewrite_turns += source.latest_turn_only_rewrite_turns
def _disable_headroom_benchmark_logging() -> None:
@ -914,6 +1095,32 @@ def _write_checkpoint_by_session_id(
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
def _update_prefix_tracker(
prefix_tracker: PrefixCacheTracker,
*,
cache_read_tokens: int,
cache_write_tokens: int,
messages: list[dict[str, Any]],
message_token_counts: list[int],
original_messages: list[dict[str, Any]] | None = None,
) -> None:
try:
prefix_tracker.update_from_response(
cache_read_tokens=cache_read_tokens,
cache_write_tokens=cache_write_tokens,
messages=messages,
message_token_counts=message_token_counts,
original_messages=original_messages,
)
except TypeError:
prefix_tracker.update_from_response(
cache_read_tokens=cache_read_tokens,
cache_write_tokens=cache_write_tokens,
messages=messages,
message_token_counts=message_token_counts,
)
def _simulate_single_replay_mode(
replay: SessionReplay,
mode: str,
@ -938,6 +1145,7 @@ def _simulate_single_replay_mode(
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(
@ -950,6 +1158,17 @@ def _simulate_single_replay_mode(
previous_original_messages=previous_original_context,
previous_forwarded_messages=previous_forwarded_context,
)
rewrite, retroactive_rewrite = _rewrite_scope(
conversation,
forwarded,
stable_prefix_message_count=prior_context_message_count,
)
if rewrite:
summary.rewrite_turns += 1
if retroactive_rewrite:
summary.retroactive_rewrite_turns += 1
else:
summary.latest_turn_only_rewrite_turns += 1
if pending is not None:
_apply_turn_metrics(
pending.summary,
@ -968,7 +1187,8 @@ def _simulate_single_replay_mode(
previous_timestamp = pending.turn.timestamp
if prefix_tracker is not None:
prefix_tracker.update_from_response(
_update_prefix_tracker(
prefix_tracker,
cache_read_tokens=0,
cache_write_tokens=0,
messages=forwarded,
@ -1231,12 +1451,60 @@ def determine_winners(summaries: dict[str, ModeSummary]) -> dict[str, str]:
}
def _metric_value(summary: ModeSummary, field: str) -> float:
value = getattr(summary, field)
return float(value)
def classify_metric_impact(
baseline: ModeSummary,
candidate: ModeSummary,
field: str,
) -> dict[str, float | str]:
baseline_value = _metric_value(baseline, field)
candidate_value = _metric_value(candidate, field)
delta = candidate_value - baseline_value
direction = IMPACT_DIRECTION[field]
tolerance = 1e-9
if abs(delta) <= tolerance:
impact = "no_change"
elif direction == "lower":
impact = "assist" if delta < 0 else "harm"
elif direction == "higher":
impact = "assist" if delta > 0 else "harm"
else:
impact = "harm" if abs(delta) > tolerance else "no_change"
return {
"baseline": baseline_value,
"candidate": candidate_value,
"delta": delta,
"impact": impact,
"direction": direction,
}
def summarize_mode_impact_vs_baseline(
summaries: dict[str, ModeSummary],
) -> dict[str, dict[str, dict[str, float | str]]]:
baseline = summaries["baseline"]
result: dict[str, dict[str, dict[str, float | str]]] = {}
for mode in (PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
candidate = summaries[mode]
result[mode] = {
field: classify_metric_impact(baseline, candidate, field) for field in IMPACT_DIRECTION
}
return result
def format_currency(value: float) -> str:
return f"${value:,.2f}"
def print_console_report(dataset: DatasetSummary, summaries: dict[str, ModeSummary]) -> None:
winners = determine_winners(summaries)
impacts = summarize_mode_impact_vs_baseline(summaries)
print("Claude session mode simulation")
print(
f"Dataset: {dataset.projects} projects, {dataset.sessions} sessions, "
@ -1245,7 +1513,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 total_cost no_cache"
"mode raw_tok cache_tok cache_read cache_write paid_in paid_out busts ttl_exp rewrite retro_rw total_cost no_cache"
)
for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
summary = summaries[mode]
@ -1254,6 +1522,7 @@ 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"{format_currency(summary.total_cost_usd):>11} "
f"{format_currency(summary.no_cache_total_cost_usd):>11}"
)
@ -1265,6 +1534,26 @@ def print_console_report(dataset: DatasetSummary, summaries: dict[str, ModeSumma
"Winner if cache read tokens do not count against window: "
f"{winners['window_without_cache_reads']}"
)
print()
print("Impact vs baseline")
for mode in (PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
impact = impacts[mode]
print(
f"{mode}: total_cost={impact['total_cost_usd']['impact']} "
f"({format_currency(impact['total_cost_usd']['delta'])}), "
f"cache_read={impact['cache_read_tokens']['impact']} "
f"({int(impact['cache_read_tokens']['delta']):,}), "
f"cache_write={impact['cache_write_tokens']['impact']} "
f"({int(impact['cache_write_tokens']['delta']):,}), "
f"paid_input={impact['regular_input_tokens']['impact']} "
f"({int(impact['regular_input_tokens']['delta']):,}), "
f"rewrite={impact['rewrite_turns']['impact']} "
f"({int(impact['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']} "
f"({int(impact['prompt_window_with_cache']['delta']):,})"
)
def print_observed_console_report(observed: ObservedSummary) -> None:
@ -1288,6 +1577,7 @@ def build_report_markdown(
summaries: dict[str, ModeSummary],
) -> str:
winners = determine_winners(summaries)
impacts = summarize_mode_impact_vs_baseline(summaries)
model_lines = "\n".join(f"- `{model}`: {count}" for model, count in dataset.models.items())
rows = []
for mode in ("baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
@ -1311,12 +1601,36 @@ def build_report_markdown(
format_currency(summary.no_cache_total_cost_usd),
f"{summary.cache_bust_turns:,}",
f"{summary.ttl_expiry_turns:,}",
f"{summary.rewrite_turns:,}",
f"{summary.retroactive_rewrite_turns:,}",
f"{summary.latest_turn_only_rewrite_turns:,}",
f"{summary.prompt_window_with_cache:,}",
f"{summary.prompt_window_without_cache_reads:,}",
]
)
+ " |"
)
impact_rows = []
for mode in (PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
for metric_key, label in (
("total_cost_usd", "Total Cost"),
("cache_read_tokens", "Cache Read Tokens"),
("cache_write_tokens", "Cache Write Tokens"),
("regular_input_tokens", "Paid Input Tokens"),
("output_tokens", "Paid Output Tokens"),
("prompt_window_with_cache", "Window With Cache"),
("prompt_window_without_cache_reads", "Window Without Cache Reads"),
("cache_bust_turns", "Cache Bust Turns"),
("rewrite_turns", "Rewrite Turns"),
("retroactive_rewrite_turns", "Retroactive Rewrite Turns"),
("latest_turn_only_rewrite_turns", "Latest-Turn-Only Rewrite Turns"),
):
impact = impacts[mode][metric_key]
delta = impact["delta"]
delta_text = format_currency(delta) if "cost" in metric_key else f"{int(delta):,}"
impact_rows.append(
f"| {mode} | {label} | {impact['impact']} | {delta_text} | {impact['direction']} |"
)
return "\n".join(
[
"# Claude Session Mode Simulation",
@ -1351,10 +1665,16 @@ 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 | 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 | Retroactive Rewrite Turns | Latest-Turn-Only Rewrite Turns | Window Tokens (Cache Counted) | Window Tokens (Cache Reads Excluded) |",
"| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
*rows,
"",
"## Impact vs Baseline",
"",
"| Mode | Metric | Classification | Delta | Better Direction |",
"| --- | --- | --- | ---: | --- |",
*impact_rows,
"",
"## Winners",
"",
f"- Total cost winner: `{winners['total_cost']}`",
@ -1372,6 +1692,7 @@ def build_report_html(
summaries: dict[str, ModeSummary],
) -> str:
winners = determine_winners(summaries)
impacts = summarize_mode_impact_vs_baseline(summaries)
model_items = "".join(
f"<li><code>{model}</code><span>{count:,}</span></li>"
for model, count in dataset.models.items()
@ -1390,12 +1711,42 @@ def build_report_html(
f"<td>{summary.output_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.retroactive_rewrite_turns:,}</td>"
f"<td>{summary.latest_turn_only_rewrite_turns:,}</td>"
f"<td>{format_currency(summary.total_cost_usd)}</td>"
f"<td>{format_currency(summary.no_cache_total_cost_usd)}</td>"
f"<td>{summary.prompt_window_with_cache:,}</td>"
f"<td>{summary.prompt_window_without_cache_reads:,}</td>"
"</tr>"
)
impact_rows = []
for mode in (PROXY_MODE_TOKEN, PROXY_MODE_CACHE):
for metric_key, label in (
("total_cost_usd", "Total Cost"),
("cache_read_tokens", "Cache Read Tokens"),
("cache_write_tokens", "Cache Write Tokens"),
("regular_input_tokens", "Paid Input Tokens"),
("output_tokens", "Paid Output Tokens"),
("prompt_window_with_cache", "Window With Cache"),
("prompt_window_without_cache_reads", "Window Without Cache Reads"),
("cache_bust_turns", "Cache Bust Turns"),
("rewrite_turns", "Rewrite Turns"),
("retroactive_rewrite_turns", "Retroactive Rewrite Turns"),
("latest_turn_only_rewrite_turns", "Latest-Turn-Only Rewrite Turns"),
):
impact = impacts[mode][metric_key]
delta = impact["delta"]
delta_text = format_currency(delta) if "cost" in metric_key else f"{int(delta):,}"
impact_rows.append(
"<tr>"
f"<td><span class='badge'>{mode}</span></td>"
f"<td>{label}</td>"
f"<td>{impact['impact']}</td>"
f"<td>{delta_text}</td>"
f"<td>{impact['direction']}</td>"
"</tr>"
)
return f"""<!doctype html>
<html lang="en">
<head>
@ -1517,7 +1868,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>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>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>
@ -1526,6 +1877,21 @@ def build_report_html(
</table>
</div>
</section>
<section class="section card">
<h2>Impact vs Baseline</h2>
<div class="table-wrap">
<table>
<thead>
<tr>
<th>Mode</th><th>Metric</th><th>Classification</th><th>Delta</th><th>Better Direction</th>
</tr>
</thead>
<tbody>
{"".join(impact_rows)}
</tbody>
</table>
</div>
</section>
</div>
</body>
</html>"""
@ -1548,6 +1914,7 @@ def write_report(
"observed": asdict(observed),
"summaries": {mode: asdict(summary) for mode, summary in summaries.items()},
"winners": determine_winners(summaries),
"impact_vs_baseline": summarize_mode_impact_vs_baseline(summaries),
}
json_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
return md_path, json_path, html_path

10
docs/benchmarks.md
View file

@ -209,6 +209,9 @@ python benchmarks/proxy_mode_benchmark.py --turns 12 --show-real-harness
# Replay local Claude Code transcripts (no API calls)
python benchmarks/claude_session_mode_benchmark.py --workers 1
# Compare two refs on the same local Claude transcript corpus
python benchmarks/claude_session_branch_compare.py --left-ref upstream/main --right-ref HEAD --recent-turns-per-session 200 --workers 1
```
This benchmark compares `token` vs `cache` proxy modes on the same synthetic conversation:
@ -218,6 +221,13 @@ This benchmark compares `token` vs `cache` proxy modes on the same synthetic con
`--show-real-harness` prints optional steps for running the same comparison with Claude Code, but does not call APIs by default.
`claude_session_branch_compare.py` runs the real local session replay benchmark twice, once per git ref, in isolated worktrees. It writes:
- per-ref replay outputs under `benchmark_results/branch_compare/<label>/`
- a combined comparison report under `benchmark_results/branch_compare/`
Use it when you want a clean PR-vs-`main` comparison on the same transcript slice.
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

106
headroom/proxy/handlers/anthropic.py
View file

@ -179,6 +179,100 @@ class AnthropicHandlerMixin:
copy.deepcopy(current_messages[prefix_len:]),
)
@staticmethod
def _extract_cache_stable_last_message_suffix(
current_messages: list[dict[str, Any]],
previous_original_messages: list[dict[str, Any]] | None,
previous_forwarded_messages: list[dict[str, Any]] | None,
) -> tuple[list[dict[str, Any]], dict[str, Any], list[dict[str, Any]]] | None:
"""Return append-only delta when only the latest message grew in place."""
if not previous_original_messages or previous_forwarded_messages is None:
return None
if (
len(current_messages) != len(previous_original_messages)
or len(previous_forwarded_messages) != len(previous_original_messages)
or not current_messages
):
return None
prefix_len = len(current_messages) - 1
if (
prefix_len > 0
and current_messages[:prefix_len] != previous_original_messages[:prefix_len]
):
return None
current_last = current_messages[-1]
previous_original_last = previous_original_messages[-1]
previous_forwarded_last = previous_forwarded_messages[-1]
if current_last.get("role") != previous_original_last.get("role") or current_last.get(
"role"
) != previous_forwarded_last.get("role"):
return None
current_content = current_last.get("content")
previous_original_content = previous_original_last.get("content")
previous_forwarded_content = previous_forwarded_last.get("content")
if (
isinstance(current_content, str)
and isinstance(previous_original_content, str)
and isinstance(previous_forwarded_content, str)
and current_content.startswith(previous_original_content)
):
suffix = current_content[len(previous_original_content) :]
delta_messages = []
if suffix:
delta_messages = [{**copy.deepcopy(current_last), "content": suffix}]
return (
copy.deepcopy(previous_forwarded_messages[:-1]),
copy.deepcopy(previous_forwarded_last),
delta_messages,
)
if (
isinstance(current_content, list)
and isinstance(previous_original_content, list)
and isinstance(previous_forwarded_content, list)
and len(current_content) >= len(previous_original_content)
and current_content[: len(previous_original_content)] == previous_original_content
):
delta_blocks = copy.deepcopy(current_content[len(previous_original_content) :])
delta_messages = []
if delta_blocks:
delta_messages = [{**copy.deepcopy(current_last), "content": delta_blocks}]
return (
copy.deepcopy(previous_forwarded_messages[:-1]),
copy.deepcopy(previous_forwarded_last),
delta_messages,
)
return None
@staticmethod
def _merge_appended_message_delta(
previous_forwarded_message: dict[str, Any],
delta_forwarded_message: dict[str, Any] | None,
) -> dict[str, Any] | None:
"""Merge a compressed suffix back into the prior forwarded message."""
if delta_forwarded_message is None:
return copy.deepcopy(previous_forwarded_message)
if previous_forwarded_message.get("role") != delta_forwarded_message.get("role"):
return None
previous_content = previous_forwarded_message.get("content")
delta_content = delta_forwarded_message.get("content")
if isinstance(previous_content, str) and isinstance(delta_content, str):
return {
**copy.deepcopy(previous_forwarded_message),
"content": previous_content + delta_content,
}
if isinstance(previous_content, list) and isinstance(delta_content, list):
return {
**copy.deepcopy(previous_forwarded_message),
"content": copy.deepcopy(previous_content) + copy.deepcopy(delta_content),
}
return None
@staticmethod
def _assistant_message_from_response_json(
resp_json: dict[str, Any] | None,
@ -482,10 +576,7 @@ class AnthropicHandlerMixin:
previous_original_messages,
previous_forwarded_messages,
)
if delta is None:
optimized_messages = messages
optimized_tokens = original_tokens
else:
if delta is not None:
stable_forwarded_prefix, delta_messages = delta
if delta_messages:
result = await asyncio.wait_for(
@ -508,6 +599,13 @@ class AnthropicHandlerMixin:
else:
optimized_messages = stable_forwarded_prefix
optimized_tokens = tokenizer.count_messages(optimized_messages)
else:
# Conservative rule for cache mode:
# only replay exact stable message-prefix extensions.
# In-message append rewriting is deferred until we can
# prove it is perfectly replayable across future turns.
optimized_messages = messages
optimized_tokens = original_tokens
if result and result.waste_signals:
waste_signals_dict = result.waste_signals.to_dict()

160
tests/test_claude_session_branch_compare.py Normal file
View file

@ -0,0 +1,160 @@
from __future__ import annotations
import json
import sys
from pathlib import Path
from benchmarks.claude_session_branch_compare import (
BranchResult,
_build_benchmark_command,
_build_six_way_rows,
_classify_delta,
_ref_slug,
build_compare_markdown,
write_compare_report,
)
def _branch(label: str, ref: str, commit: str, total_cost: float) -> BranchResult:
summaries = {
"baseline": {
"mode": "baseline",
"total_cost_usd": total_cost + 1.0,
"no_cache_total_cost_usd": total_cost + 5.0,
"forwarded_input_tokens": 1_200,
"cache_read_tokens": 800,
"cache_write_tokens": 200,
"regular_input_tokens": 400,
"output_tokens": 120,
"cache_bust_turns": 1,
"ttl_expiry_turns": 2,
"prompt_window_with_cache": 1_200,
"prompt_window_without_cache_reads": 400,
},
"token": {
"mode": "token",
"total_cost_usd": total_cost,
"no_cache_total_cost_usd": total_cost + 3.0,
"forwarded_input_tokens": 900,
"cache_read_tokens": 700,
"cache_write_tokens": 150,
"regular_input_tokens": 200,
"output_tokens": 120,
"cache_bust_turns": 4,
"ttl_expiry_turns": 2,
"prompt_window_with_cache": 900,
"prompt_window_without_cache_reads": 200,
},
"cache": {
"mode": "cache",
"total_cost_usd": total_cost + 0.5,
"no_cache_total_cost_usd": total_cost + 4.0,
"forwarded_input_tokens": 950,
"cache_read_tokens": 760,
"cache_write_tokens": 180,
"regular_input_tokens": 190,
"output_tokens": 120,
"cache_bust_turns": 1,
"ttl_expiry_turns": 2,
"prompt_window_with_cache": 950,
"prompt_window_without_cache_reads": 190,
},
}
return BranchResult(
ref=ref,
label=label,
commit=commit,
summary=f"{label} summary",
dataset={
"projects": 3,
"sessions": 7,
"requests": 80,
"sampled_requests": 80,
"sampling_note": "Most recent 10 turns per session",
},
observed={"cache_ratio_pct": 97.0},
summaries=summaries,
winners={
"total_cost": "token",
"no_cache_total_cost": "token",
"window_with_cache": "token",
"window_without_cache_reads": "cache",
},
output_dir=f"benchmark_results/{label}",
)
def test_ref_slug_normalizes_refs() -> None:
assert _ref_slug("upstream/main") == "upstream-main"
assert _ref_slug("feature/cache.fix") == "feature-cache-fix"
def test_build_benchmark_command_includes_knobs() -> None:
command = _build_benchmark_command(
python_executable=sys.executable,
script_path=Path("benchmarks") / "claude_session_mode_benchmark.py",
root=Path.home() / ".claude" / "projects",
output_dir=Path("benchmark_results") / "pr",
max_sessions=5,
recent_turns_per_session=200,
cache_ttl_minutes=5,
cache_write_multiplier=1.25,
workers=1,
)
assert command[0] == sys.executable
assert "--max-sessions" in command
assert "--recent-turns-per-session" in command
assert "--workers" in command
def test_build_compare_markdown_surfaces_branch_deltas() -> None:
left = _branch("main", "upstream/main", "abc123456789", 12.0)
right = _branch("pr", "HEAD", "def987654321", 11.0)
markdown = build_compare_markdown(left, right)
assert "Claude Session Branch Comparison" in markdown
assert "`main`" not in markdown
assert "main picks" not in markdown
assert "Delta (pr - main)" in markdown
assert "| token | Total Cost | $12.00 | $11.00 | $-1.00 |" in markdown
def test_write_compare_report_persists_payload(tmp_path: Path) -> None:
left = _branch("main", "upstream/main", "abc123456789", 12.0)
right = _branch("pr", "HEAD", "def987654321", 11.0)
md_path, json_path, html_path = write_compare_report(tmp_path, left, right)
assert md_path.exists()
assert html_path.exists()
payload = json.loads(json_path.read_text(encoding="utf-8"))
assert payload["left"]["ref"] == "upstream/main"
assert payload["right"]["label"] == "pr"
assert payload["right_winners"]["total_cost"] == "token"
def test_branch_delta_classification_uses_metric_direction() -> None:
assert _classify_delta("total_cost_usd", -1.0) == "assist"
assert _classify_delta("cache_read_tokens", 10.0) == "assist"
assert _classify_delta("cache_write_tokens", 5.0) == "harm"
assert _classify_delta("output_tokens", 0.0) == "no_change"
def test_six_way_rows_cover_both_branches_and_modes() -> None:
left = _branch("main", "upstream/main", "abc123456789", 12.0)
right = _branch("pr", "HEAD", "def987654321", 11.0)
rows = _build_six_way_rows(left, right)
assert len(rows) == 6
assert rows[0]["branch"] == "main"
assert rows[0]["mode"] == "baseline"
assert any(row["branch"] == "pr" and row["mode"] == "token" for row in rows)
assert any(
row["branch"] == "main"
and row["mode"] == "token"
and row["paid_input_delta_vs_branch_baseline"] == -200
for row in rows
)

188
tests/test_claude_session_mode_benchmark.py
View file

@ -5,6 +5,7 @@ from __future__ import annotations
import json
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from benchmarks.claude_session_mode_benchmark import (
PROXY_MODE_CACHE,
@ -12,13 +13,18 @@ from benchmarks.claude_session_mode_benchmark import (
ModeSummary,
ReplayTurn,
SessionReplay,
_extract_cache_stable_last_message_suffix,
_merge_appended_message_delta,
_rewrite_scope,
_write_checkpoint_by_session_id,
build_dataset_and_observed_from_files,
classify_metric_impact,
decode_project_key,
determine_winners,
load_session_replay,
resolve_checkpoint_dir,
simulate_replays,
summarize_mode_impact_vs_baseline,
summarize_observed_usage,
trim_replay_to_recent_turns,
)
@ -147,6 +153,8 @@ def test_simulation_and_winner_logic() -> None:
assert summaries["baseline"].cache_bust_turns == 0
assert summaries[PROXY_MODE_CACHE].cache_bust_turns == 0
assert summaries[PROXY_MODE_TOKEN].cache_bust_turns >= 0
assert summaries[PROXY_MODE_TOKEN].rewrite_turns >= 0
assert summaries[PROXY_MODE_CACHE].rewrite_turns >= 0
winners = determine_winners(summaries)
assert winners["total_cost"] in {"baseline", PROXY_MODE_TOKEN, PROXY_MODE_CACHE}
@ -334,3 +342,183 @@ def test_resolve_checkpoint_dir_namespaces_sampling_mode() -> None:
assert (
resolve_checkpoint_dir(base, recent_turns_per_session=200).name == "v4__ttl_5m__recent_200"
)
def test_cache_suffix_helpers_support_append_only_text_growth() -> None:
suffix_delta = _extract_cache_stable_last_message_suffix(
[{"role": "user", "content": "prefix + raw suffix"}],
[{"role": "user", "content": "prefix"}],
[{"role": "user", "content": "COMPRESSED_PREFIX"}],
)
assert suffix_delta is not None
stable_prefix, stable_last_message, delta_messages = suffix_delta
assert stable_prefix == []
assert stable_last_message == {"role": "user", "content": "COMPRESSED_PREFIX"}
assert delta_messages == [{"role": "user", "content": " + raw suffix"}]
merged = _merge_appended_message_delta(
stable_last_message,
{"role": "user", "content": " + COMPRESSED_SUFFIX"},
)
assert merged == {"role": "user", "content": "COMPRESSED_PREFIX + COMPRESSED_SUFFIX"}
def test_mode_impact_classification_marks_assist_harm_and_no_change() -> None:
baseline = ModeSummary(
mode="baseline",
forwarded_input_tokens=100,
cache_read_tokens=50,
cache_write_tokens=10,
regular_input_tokens=40,
output_tokens=5,
total_cost_usd=1.0,
)
token = ModeSummary(
mode=PROXY_MODE_TOKEN,
forwarded_input_tokens=80,
cache_read_tokens=70,
cache_write_tokens=8,
regular_input_tokens=30,
output_tokens=5,
total_cost_usd=0.8,
)
cache = ModeSummary(
mode=PROXY_MODE_CACHE,
forwarded_input_tokens=120,
cache_read_tokens=45,
cache_write_tokens=15,
regular_input_tokens=60,
output_tokens=5,
total_cost_usd=1.2,
)
assert classify_metric_impact(baseline, token, "forwarded_input_tokens")["impact"] == "assist"
assert classify_metric_impact(baseline, token, "cache_read_tokens")["impact"] == "assist"
assert classify_metric_impact(baseline, cache, "total_cost_usd")["impact"] == "harm"
assert classify_metric_impact(baseline, token, "output_tokens")["impact"] == "no_change"
impacts = summarize_mode_impact_vs_baseline(
{"baseline": baseline, PROXY_MODE_TOKEN: token, PROXY_MODE_CACHE: cache}
)
assert impacts[PROXY_MODE_TOKEN]["total_cost_usd"]["impact"] == "assist"
assert impacts[PROXY_MODE_CACHE]["cache_write_tokens"]["impact"] == "harm"
def test_rewrite_scope_distinguishes_retroactive_from_latest_turn_only() -> None:
rewrite, retroactive = _rewrite_scope(
[{"role": "user", "content": "prefix"}, {"role": "user", "content": "new raw"}],
[{"role": "user", "content": "prefix"}, {"role": "user", "content": "new compressed"}],
stable_prefix_message_count=1,
)
assert rewrite is True
assert retroactive is False
rewrite, retroactive = _rewrite_scope(
[{"role": "user", "content": "prefix"}, {"role": "user", "content": "new raw"}],
[
{"role": "user", "content": "compressed prefix"},
{"role": "user", "content": "new compressed"},
],
stable_prefix_message_count=1,
)
assert rewrite is True
assert retroactive is True
def test_synthetic_token_mode_busts_cache_while_cache_mode_stays_stable(monkeypatch) -> None:
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(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()
monkeypatch.setattr(
"benchmarks.claude_session_mode_benchmark._make_proxy",
lambda mode: _FakeProxy(),
)
tool_blob = "X" * 800
replay = SessionReplay(
session_id="synth-bust",
project_key="C--git-synth",
decoded_project_path=r"C:\git\synth",
turns=[
ReplayTurn(
session_id="synth-bust",
project_key="C--git-synth",
decoded_project_path=r"C:\git\synth",
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": tool_blob,
}
],
},
],
assistant_message={"role": "assistant", "content": "ok"},
output_tokens=10,
),
ReplayTurn(
session_id="synth-bust",
project_key="C--git-synth",
decoded_project_path=r"C:\git\synth",
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,
),
],
)
_, summaries = simulate_replays([replay], cache_ttl_minutes=5)
token = summaries[PROXY_MODE_TOKEN]
cache = summaries[PROXY_MODE_CACHE]
assert token.cache_bust_turns == 1
assert token.rewrite_turns >= 1
assert token.retroactive_rewrite_turns >= 1
assert cache.cache_bust_turns == 0
assert cache.retroactive_rewrite_turns == 0

75
tests/test_proxy_anthropic_cache_stability.py
View file

@ -897,3 +897,78 @@ def test_cache_mode_reuses_prior_forwarded_prefix_and_compresses_only_new_suffix
{"role": "assistant", "content": "turn2-assistant"},
{"role": "user", "content": "COMPRESSED_TURN3"},
]
def test_cache_mode_skips_same_message_append_rewrite_to_preserve_stability() -> None:
captured = {"calls": []}
with _make_proxy_client() as client:
proxy = client.app.state.proxy
proxy.config.optimize = True
proxy.config.mode = "cache"
proxy.config.image_optimize = False
tracker = _FakePrefixTracker(frozen_count=0)
tracker._last_original_messages = [
{"role": "user", "content": "shared-prefix"},
]
tracker._last_forwarded_messages = [
{"role": "user", "content": "COMPRESSED_PREFIX"},
]
tracker.get_last_original_messages = lambda: tracker._last_original_messages.copy()
tracker.get_last_forwarded_messages = lambda: tracker._last_forwarded_messages.copy()
proxy.session_tracker_store.compute_session_id = lambda request, model, messages: (
"stable-session"
)
proxy.session_tracker_store.get_or_create = lambda session_id, provider: tracker
def _fake_apply(**kwargs):
captured["calls"].append(kwargs["messages"])
return SimpleNamespace(
messages=[{"role": "user", "content": " + COMPRESSED_SUFFIX"}],
transforms_applied=["fake:suffix"],
timing={},
tokens_before=20,
tokens_after=10,
waste_signals=None,
)
proxy.anthropic_pipeline.apply = _fake_apply
async def _fake_retry(method, url, headers, body, stream=False): # noqa: ANN001
captured["body"] = body
return httpx.Response(
200,
json={
"id": "msg_cache_suffix",
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "ok"}],
"usage": {
"input_tokens": 80,
"output_tokens": 3,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0,
},
},
)
proxy._retry_request = _fake_retry
response = client.post(
"/v1/messages",
headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"},
json={
"model": "claude-sonnet-4-6",
"max_tokens": 64,
"messages": [
{"role": "user", "content": "shared-prefix + raw suffix"},
],
},
)
assert response.status_code == 200
assert captured["calls"] == []
assert captured["body"]["messages"] == [
{"role": "user", "content": "shared-prefix + raw suffix"},
]