Skip to main content

lean_ctx/proxy/
anthropic.rs

1use axum::{
2    body::Body,
3    extract::State,
4    http::{Request, StatusCode},
5    response::Response,
6};
7use serde_json::Value;
8
9use super::ProxyState;
10use super::compress_shared::{self, ToolKind};
11use super::forward;
12use super::tool_kind::{self, ToolResultKind};
13use super::{cache_safety, prefix_cache_stats, prefix_replay, prose, sticky_tools};
14
15std::thread_local! {
16    /// Set by `forward.rs` when the current request has the `X-Headroom-Compressed` header.
17    static HEADROOM_REQUEST: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
18}
19
20/// Called from `forward.rs` to signal that this request was pre-compressed by Headroom.
21pub(super) fn set_headroom_request(val: bool) {
22    HEADROOM_REQUEST.set(val);
23}
24use crate::core::config::{HistoryMode, ProseRole};
25
26pub async fn handler(
27    State(state): State<ProxyState>,
28    req: Request<Body>,
29) -> Result<Response, StatusCode> {
30    let upstream = state.anthropic_upstream();
31    forward::forward_request(
32        State(state),
33        req,
34        &upstream,
35        "/v1/messages",
36        compress_request_body,
37        "Anthropic",
38        &[],
39    )
40    .await
41}
42
43pub(super) fn compress_request_body(
44    parsed: Value,
45    original_size: usize,
46) -> (Vec<u8>, usize, usize) {
47    let mut doc = parsed;
48    let mut modified = false;
49
50    // Opt-in per-role prose aggressiveness (#710). Both default to `None`, in
51    // which case nothing below fires and the body is byte-for-byte unchanged.
52    let cfg = crate::core::config::Config::load();
53    let config_headroom = cfg.proxy.is_headroom_compat();
54    if config_headroom {
55        prefix_cache_stats::record_headroom_compat();
56    }
57    // Per-request Headroom detection: if this specific request carries the
58    // X-Headroom-Compressed header, treat it as headroom-compat even without
59    // the global config flag. The header is checked in forward.rs and the
60    // result is threaded through via a thread-local set by the caller.
61    let _headroom_compat = config_headroom || HEADROOM_REQUEST.get();
62    let system_aggr = cfg.proxy.resolved_role_aggressiveness(ProseRole::System);
63    let user_aggr = cfg.proxy.resolved_role_aggressiveness(ProseRole::User);
64    let live_compress = cfg.proxy.live_compresses();
65    let mode = cfg.proxy.resolved_history_mode();
66    // #939: active prompt-cache breakpoint injection (opt-in, Anthropic-only).
67    // Resolved up front so the meter-only short-circuit below does not skip the
68    // one mutation this mode performs — its whole point is to add a cache anchor
69    // to an otherwise byte-passthrough request.
70    let inject_breakpoint = cfg.proxy.cache_breakpoint_enabled();
71    // #940: cache-aligner volatile-field telemetry (default-on, measurement-only).
72    // Also resolved up front so a meter-only proxy still reaches the scan slot —
73    // it never mutates the body, it only records how much cache the system prompt
74    // leaks, so it ships on for every proxy (#986 premium defaults).
75    let align_volatile = cfg.proxy.cache_aligner_enabled();
76    // #974: active cache-aligner relocate (opt-in, Anthropic-only). Resolved up
77    // front like the telemetry above so a meter-only proxy still reaches the
78    // relocate slot — this is the one mutation that moves volatile fields out of
79    // the cacheable prefix.
80    let relocate_volatile = cfg.proxy.cache_align_relocate_enabled();
81    // #986: cache-economics (default-on). Resolved up front so the meter-only
82    // short-circuit below still reaches the miss-attribution slot — that
83    // telemetry only reads the cacheable prefix, it never mutates the body, and
84    // the paired net-cost gate only makes the cold-prefix repack more
85    // conservative, so both halves ship on for every proxy (#986 premium
86    // defaults).
87    let cache_economics = cfg.proxy.cache_policy_enabled();
88    // #895 Track B: output-savings holdout arm, from the pristine body (before any
89    // mutation below) so it matches the arm the response meter records. Control
90    // conversations skip output-shaping (effort + verbosity steer) but are still
91    // metered. Default holdout=0 → always Treatment (no behaviour change).
92    let arm = super::holdout::assign(
93        &super::holdout::anthropic_key(&doc),
94        cfg.proxy.output_holdout_fraction(),
95    );
96    // #493: in-band CCR expansion (opt-in). Splice any <lc_expand:HASH> the model
97    // echoed back into the verbatim original from the local tee store. A strict
98    // no-op when no marker is present (byte-identical body → cache-safe). Runs
99    // before the meter-only short-circuit so an explicit expand request is
100    // honored even when the proxy is otherwise byte-passthrough.
101    if cfg.proxy.ccr_inband_enabled() {
102        modified |= super::ccr::splice_inband_in_place(&mut doc);
103    }
104    // #834: cache-safe cross-provider effort control. Default off → no-op. The
105    // value is a constant, so it never perturbs the prompt-cache prefix; it only
106    // dials an *existing* adaptive thinking request (never enables thinking the
107    // client didn't ask for).
108    if arm == super::holdout::Arm::Treatment {
109        if let Some(effort) = cfg.proxy.resolved_effort() {
110            modified |= super::effort::apply_anthropic(&mut doc, effort);
111        }
112        // #895: cache-safe wire verbosity steer (constant suffix after the last
113        // cache_control breakpoint). Control arm skips it so the holdout measures
114        // its effect.
115        if cfg.proxy.verbosity_steer_enabled() {
116            modified |= super::verbosity::apply_anthropic(&mut doc);
117        }
118    }
119    // Meter-only (#481): live compression off, no history pruning, no prose
120    // rewriting → forward + usage metering still run, but the body is left
121    // unchanged so the provider prompt-cache prefix stays byte-stable. A pending
122    // in-band splice (`modified`) opts out: the body did change this turn.
123    if !live_compress
124        && mode == HistoryMode::Off
125        && system_aggr.is_none()
126        && user_aggr.is_none()
127        && !modified
128        && !inject_breakpoint
129        && !align_volatile
130        && !relocate_volatile
131        && !cache_economics
132    {
133        let out = serde_json::to_vec(&doc).unwrap_or_default();
134        return (out, original_size, original_size);
135    }
136    let mut prose_segments: u64 = 0;
137
138    // Length of the client's provider-cached message prefix. Needed both for
139    // cache-safe pruning below and to gate top-level system prose: if any
140    // message is client-cached, `system` (which precedes every message) is part
141    // of that cached prefix and must not be rewritten.
142    let cached = doc
143        .get("messages")
144        .and_then(|m| m.as_array())
145        .map_or(0, |m| super::history_prune::cached_prefix_len(m));
146
147    // #480: opt-in big-gap cold-prefix repack. When enabled AND the proxy can
148    // confidently predict (from idle time vs the provider cache TTL) that the
149    // client-cached prefix is already cold, override the normal "never touch the
150    // cached prefix" rule for THIS request and prune/compress the prefix too,
151    // re-seeding a leaner cache. Default-off; never fires without a measured idle
152    // gap past TTL × margin, so warm caches stay byte-stable (#448).
153    // #986: cache-economics miss attribution (opt-in, measurement-only). Classify
154    // why this turn hits or misses the provider prompt-cache (TTL lapse vs prefix
155    // change) and bump the `/status` gauges. Reads the cacheable prefix only — the
156    // body is never touched — so it is strictly cache-safe.
157    if cache_economics
158        && let Some(m) = doc.get("messages").and_then(|m| m.as_array())
159        && let Some(outcome) = super::cache_attribution::record_request(m, cached)
160    {
161        match outcome {
162            super::cache_attribution::CacheOutcome::WarmReuse => {
163                prefix_cache_stats::record_hit();
164            }
165            super::cache_attribution::CacheOutcome::ColdStart => {}
166            _ => {
167                prefix_cache_stats::record_miss();
168            }
169        }
170    }
171    // #480 repack decision, with the #986 net-cost gate folded in: when
172    // cache-economics is on, also require the prefix to be large enough to cache
173    // (`worth_repacking`). The gate is an extra AND-condition, so it can only make
174    // repacking *more* conservative; default-off proxies keep the prior value.
175    let repack = cfg.proxy.repacks_cold_prefix()
176        && doc
177            .get("messages")
178            .and_then(|m| m.as_array())
179            .is_some_and(|m| {
180                super::cold_prefix::repack_decision(m, cached)
181                    && (!cache_economics
182                        || super::cache_policy::worth_repacking(doc.get("system"), m, cached))
183            });
184    // The prefix length the rewrites below must protect: the full cached prefix
185    // normally, or 0 when we are intentionally repacking the cold prefix.
186    let protect = if repack { 0 } else { cached };
187
188    // System prose: only when nothing is client-cached and the `system` field
189    // carries no `cache_control` of its own — otherwise it anchors the cache.
190    // A cold-prefix repack (`protect == 0` with `repack`) deliberately rewrites
191    // it to re-seed a leaner cache.
192    let model_name = doc
193        .get("model")
194        .and_then(Value::as_str)
195        .unwrap_or("default")
196        .to_owned();
197    if let Some(a) = system_aggr
198        && protect == 0
199        && let Some(system) = doc.get_mut("system")
200        && (repack || !prose::value_has_cache_control(system))
201    {
202        let should_compress = if repack || cached == 0 {
203            true
204        } else {
205            let sys_tokens = prose::estimate_tokens(system) as u64;
206            let estimated_after = (sys_tokens as f64 * (1.0 - a)).max(0.0) as u64;
207            let reuse_rate = super::cache_attribution::estimated_reuse_rate();
208            let model_cost = super::cache_policy::model_cost_for(&model_name);
209            let gate = super::cache_policy::should_mutate_frozen(
210                sys_tokens,
211                estimated_after,
212                reuse_rate,
213                &model_cost,
214            );
215            matches!(gate, super::cache_policy::MutationDecision::Mutate { .. })
216        };
217        if should_compress {
218            let n = prose::compress_system_value(system, a);
219            if n > 0 {
220                prose_segments += u64::from(n);
221                modified = true;
222            }
223        }
224    }
225
226    if let Some(messages) = doc.get_mut("messages").and_then(|m| m.as_array_mut()) {
227        // Resolve tool-call id → tool name so file/source reads can be protected
228        // from lossy compression that would force the model to re-read mid-task.
229        let tool_names = tool_kind::anthropic_tool_names(messages);
230
231        // Prune at a frozen, cache-aware boundary by default: Anthropic's
232        // prompt cache matches exact prefixes, so the boundary must not move
233        // every turn (see `history_prune::prune_boundary`). `mode` resolved above.
234        let boundary = super::history_prune::prune_boundary(mode, messages.len());
235        // Never rewrite content the client has marked with `cache_control`:
236        // pruning inside the already-cached prefix invalidates Anthropic's
237        // prompt cache from the first changed message (#448). Pruning therefore
238        // starts after the last breakpoint; with no breakpoint this is 0, i.e.
239        // the previous behaviour.
240        modified |=
241            super::history_prune::prune_history_range(messages, protect, boundary, &tool_names);
242
243        for msg in messages.iter_mut() {
244            let role = msg.get("role").and_then(|r| r.as_str()).unwrap_or("");
245            if role != "user" {
246                continue;
247            }
248
249            if let Some(content) = msg.get_mut("content").and_then(|c| c.as_array_mut()) {
250                for block in content.iter_mut() {
251                    if compress_shared::classify_tool_kind(block) != ToolKind::ToolResult {
252                        continue;
253                    }
254
255                    let name = block
256                        .get("tool_use_id")
257                        .and_then(|v| v.as_str())
258                        .and_then(|id| tool_names.get(id))
259                        .map(String::as_str);
260                    let kind = compress_shared::tool_result_kind(name);
261
262                    // #481: skip live compression when globally off or when the
263                    // originating tool is on the exclusion list (Serena default).
264                    let excluded =
265                        name.is_some_and(|n| cfg.proxy.is_tool_live_compress_excluded(n));
266                    if live_compress
267                        && !excluded
268                        && let Some(inner_content) = block.get_mut("content")
269                    {
270                        modified |= compress_content_field(inner_content, name, kind);
271                    }
272                }
273            }
274        }
275
276        // Frozen-region user prose: free-text `text` blocks of user turns in
277        // `[cached, boundary)`. Cache-safe by construction — the cached prefix
278        // and the live tail (`>= boundary`) are both left intact, and the
279        // rewrite is content-deterministic so the prefix stays byte-stable.
280        if let Some(a) = user_aggr {
281            let end = boundary.min(messages.len());
282            let start = protect.min(end);
283            for msg in &mut messages[start..end] {
284                if msg.get("role").and_then(|r| r.as_str()) == Some("user")
285                    && let Some(content) = msg.get_mut("content").and_then(|c| c.as_array_mut())
286                {
287                    prose_segments += u64::from(prose::compress_text_blocks(content, a));
288                }
289            }
290        }
291    }
292
293    if prose_segments > 0 {
294        modified = true;
295    }
296    // #940: cache-aligner telemetry. On an unanchored system prompt (the prefix a
297    // provider would cache), count the volatile fields that would bust that cache
298    // turn-to-turn. Pure measurement — runs before any breakpoint injection and
299    // never mutates the body — so it is strictly cache-safe. Skipped once the
300    // client has anchored the prefix itself.
301    if align_volatile
302        && cached == 0
303        && let Some(system) = doc.get("system")
304        && !prose::value_has_cache_control(system)
305        && let Some(text) = super::cache_aligner::system_text(system)
306    {
307        let scan = super::cache_aligner::scan_volatile(&text);
308        cache_safety::record_volatile_system(scan.fields as u64);
309    }
310    // #974: active cache-aligner relocate (Anthropic-only). After the telemetry
311    // above measured the leak on the pristine prompt, move the volatile values out
312    // of the cacheable prefix into an uncached tail block so the prefix finally
313    // caches. Gated exactly like the breakpoint below — Treatment arm, and only
314    // when the client anchored nothing of its own. The rewrite adds the
315    // `cache_control` itself, so a following #939 injection sees an anchored prefix
316    // and stays a no-op: the two compose to exactly one breakpoint on the stable
317    // block, with the volatile tail left uncached.
318    if relocate_volatile
319        && arm == super::holdout::Arm::Treatment
320        && cached == 0
321        && doc
322            .get("system")
323            .is_some_and(|s| !prose::value_has_cache_control(s))
324    {
325        let relocated = super::cache_aligner::apply_anthropic_relocate(&mut doc);
326        if relocated > 0 {
327            modified = true;
328            cache_safety::record_volatile_relocated(relocated as u64);
329        }
330    }
331    // #939: active prompt-cache breakpoint injection (Anthropic-only). When the
332    // client anchored no prefix of its own — no message `cache_control` (`cached
333    // == 0`) and no breakpoint already on `system` — add one ephemeral breakpoint
334    // to `system` so the large, stable system prompt bills later turns at the
335    // cached rate (the win a raw API client leaves on the table). Runs after every
336    // frozen-region rewrite so the marker anchors the final system bytes and the
337    // prefix it creates stays byte-stable across turns (#498). Counted on its own
338    // gauge — a pure win, never against the cache-safe ratio.
339    if inject_breakpoint
340        && cached == 0
341        && doc
342            .get("system")
343            .is_some_and(|s| !prose::value_has_cache_control(s))
344        && super::cache_breakpoint::inject_anthropic_system(&mut doc)
345    {
346        modified = true;
347        cache_safety::record_breakpoint_injected();
348    }
349    // A deliberate cold-prefix repack (#480) is the one sanctioned exception to
350    // the frozen-window rule; count it on its own gauge so it never dilutes the
351    // cache-safe ratio (which exists to catch *accidental* #448 regressions).
352    // Every other rewrite lands strictly inside the cache-safe frozen window.
353    if repack {
354        cache_safety::record_cold_repack();
355    }
356    cache_safety::record(prose_segments, true);
357
358    // Sticky CCR tool injection: once a conversation has used CCR, keep
359    // ctx_expand in tools[] to avoid prefix-cache-busting tool-list changes.
360    let system_val = doc.get("system");
361    let messages_for_id = doc.get("messages").and_then(Value::as_array);
362    if let Some(msgs) = messages_for_id {
363        let conv_id = prefix_replay::conversation_id(system_val, msgs);
364        if sticky_tools::ensure_tool_present(conv_id, &mut doc) {
365            modified = true;
366            prefix_cache_stats::record_sticky_injection();
367        }
368    }
369
370    prefix_cache_stats::record_frozen_count(cached as u64);
371
372    // Prefix replay: if this is an append-only turn, overlay the cached
373    // forwarded prefix bytes with the fresh delta for byte-identical prefix.
374    let system_val_replay = doc.get("system");
375    let msgs_replay = doc.get("messages").and_then(Value::as_array);
376    let out = if let Some(msgs) = msgs_replay {
377        let conv_id = prefix_replay::conversation_id(system_val_replay, msgs);
378        if let Some(delta) = prefix_replay::detect_append_only(conv_id, msgs) {
379            let delta_msgs = &msgs[delta.delta_start..];
380            if let Some(replayed) = prefix_replay::overlay_prefix(&delta.prefix_bytes, delta_msgs) {
381                prefix_cache_stats::record_replay_hit();
382                let original_bytes = serde_json::to_vec(&doc).unwrap_or_default();
383                prefix_cache_stats::record_delta(
384                    original_bytes.len() as u64,
385                    replayed.len() as u64,
386                );
387                replayed
388            } else {
389                prefix_cache_stats::record_replay_miss();
390                serde_json::to_vec(&doc).unwrap_or_default()
391            }
392        } else {
393            prefix_cache_stats::record_replay_miss();
394            serde_json::to_vec(&doc).unwrap_or_default()
395        }
396    } else {
397        serde_json::to_vec(&doc).unwrap_or_default()
398    };
399    let compressed_size = if modified { out.len() } else { original_size };
400    (out, original_size, compressed_size)
401}
402
403/// Compresses a tool_result `content` field unless it is a protected file/source
404/// read, which must reach the model intact (it is what gets edited).
405fn compress_content_field(
406    content: &mut Value,
407    tool_name: Option<&str>,
408    kind: ToolResultKind,
409) -> bool {
410    match content {
411        Value::String(s) => super::tool_output::compress_text(s, tool_name, kind),
412        Value::Array(arr) => {
413            let mut modified = false;
414            for item in arr.iter_mut() {
415                if compress_shared::should_compress_content(compress_shared::classify_tool_kind(
416                    item,
417                )) && let Some(Value::String(text)) = item.get_mut("text")
418                {
419                    modified |= super::tool_output::compress_text(text, tool_name, kind);
420                }
421            }
422            modified
423        }
424        _ => false,
425    }
426}
427
428#[cfg(test)]
429mod tests {
430    use super::super::compress::compress_tool_result;
431    use super::*;
432
433    fn source_file_body() -> Vec<u8> {
434        let code = (0..60)
435            .map(|i| format!("    let binding_{i} = compute_value_{i}(context, options);"))
436            .collect::<Vec<_>>()
437            .join("\n");
438        let body = serde_json::json!({
439            "model": "claude-opus-4-8",
440            "messages": [
441                {
442                    "role": "assistant",
443                    "content": [{"type": "tool_use", "id": "toolu_1", "name": "Read", "input": {"file_path": "src/app.rs"}}]
444                },
445                {
446                    "role": "user",
447                    "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": code}]
448                }
449            ]
450        });
451        serde_json::to_vec(&body).unwrap()
452    }
453
454    #[test]
455    fn read_tool_result_is_never_truncated() {
456        let bytes = source_file_body();
457        let body: Value = serde_json::from_slice(&bytes).unwrap();
458        let (out, _orig, _comp) = compress_request_body(body, bytes.len());
459        let parsed: Value = serde_json::from_slice(&out).unwrap();
460        let content = parsed["messages"][1]["content"][0]["content"]
461            .as_str()
462            .unwrap();
463        assert!(
464            content.contains("binding_59"),
465            "the full source body must survive — refactors need it intact"
466        );
467        assert!(!content.contains("lines omitted"));
468    }
469
470    fn forge_log_body(tool_name: &str) -> Value {
471        // Generic, highly-repetitive log with no `$ cmd` hint, so routing falls
472        // back to the tool name (exercising the foreign-tool classification)
473        // and the generic compressor (not a command-specific pattern).
474        let mut log = String::new();
475        for i in 0..90 {
476            log.push_str(&format!(
477                "INFO  processing item {i}: ok, latency={i}ms, queue depth normal, retries 0\n"
478            ));
479        }
480        serde_json::json!({
481            "messages": [
482                {"role": "assistant", "content": [{"type": "tool_use", "id": "f1", "name": tool_name, "input": {}}]},
483                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "f1", "content": log}]}
484            ]
485        })
486    }
487
488    #[test]
489    fn forge_shell_tool_result_compresses() {
490        // A vendor-prefixed foreign shell tool reaches the proxy; its log output
491        // must still be compressed (rtk/ctx_* never see another server's tools).
492        let body = forge_log_body("forge_shell");
493        let bytes = serde_json::to_vec(&body).unwrap();
494        let (_out, orig, comp) = compress_request_body(body, bytes.len());
495        assert!(comp < orig, "foreign shell output must be compressed");
496    }
497
498    #[test]
499    fn foreign_read_tool_protects_source() {
500        // `forge_read` is classified FileRead via the segment fallback, so the
501        // source body must reach the model intact (it is what gets edited).
502        let code = (0..60)
503            .map(|i| format!("    let binding_{i} = compute_value_{i}(context, options);"))
504            .collect::<Vec<_>>()
505            .join("\n");
506        let body = serde_json::json!({
507            "messages": [
508                {"role": "assistant", "content": [{"type": "tool_use", "id": "r1", "name": "forge_read", "input": {"path": "src/app.rs"}}]},
509                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "r1", "content": code}]}
510            ]
511        });
512        let bytes = serde_json::to_vec(&body).unwrap();
513        let (out, _orig, _comp) = compress_request_body(body, bytes.len());
514        let parsed: Value = serde_json::from_slice(&out).unwrap();
515        let content = parsed["messages"][1]["content"][0]["content"]
516            .as_str()
517            .unwrap();
518        assert!(
519            content.contains("binding_59"),
520            "source body must survive intact"
521        );
522    }
523
524    #[test]
525    fn compress_request_body_is_deterministic() {
526        // tee path depends on the data dir; serialize env access so a parallel
527        // test never swaps LEAN_CTX_DATA_DIR between the two compressions.
528        let _lock = crate::core::data_dir::test_env_lock();
529        // #498: the proxy rewrite must be a pure function of the body so the
530        // provider prompt-cache prefix stays byte-identical across turns.
531        let bytes = serde_json::to_vec(&forge_log_body("Bash")).unwrap();
532        let a = compress_request_body(serde_json::from_slice(&bytes).unwrap(), bytes.len()).0;
533        let b = compress_request_body(serde_json::from_slice(&bytes).unwrap(), bytes.len()).0;
534        assert_eq!(a, b, "identical input must yield byte-identical output");
535    }
536
537    /// A large, highly-compressible foreign log so the live path tees + stubs it.
538    fn big_log() -> String {
539        (0..200)
540            .map(|i| format!("[info] processed item {i:04} ok, latency {i}ms, queue normal"))
541            .collect::<Vec<_>>()
542            .join("\n")
543    }
544
545    #[test]
546    fn inband_ccr_emit_echo_splice_round_trip() {
547        // Full #493 cycle through the real Anthropic request path: a lossy stub
548        // emits an <lc_expand:HASH> marker, the model echoes it, and the proxy
549        // splices the verbatim original back inline on the next request.
550        let _iso = crate::core::data_dir::isolated_data_dir();
551        crate::test_env::remove_var("LEAN_CTX_PROXY_CCR_INBAND");
552        crate::core::config::Config::update_global(|c| {
553            c.proxy.ccr_inband = Some(true);
554        })
555        .unwrap();
556
557        // EMIT: live-compress a foreign tool_result → recovery stub with a marker.
558        let log = big_log();
559        let emit = serde_json::json!({
560            "messages": [
561                {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "bash", "input": {}}]},
562                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": log}]}
563            ]
564        });
565        let bytes = serde_json::to_vec(&emit).unwrap();
566        let (out, _o, _c) = compress_request_body(emit, bytes.len());
567        let emitted: Value = serde_json::from_slice(&out).unwrap();
568        let stub = emitted["messages"][1]["content"][0]["content"]
569            .as_str()
570            .unwrap();
571        assert!(
572            stub.contains("<lc_expand:"),
573            "in-band stub must advertise an echo-able marker: {stub}"
574        );
575        assert!(
576            !stub.contains("/tee/proxy_"),
577            "in-band stub must not leak the unreachable local tee path: {stub}"
578        );
579
580        // The marker the model would copy into its next turn.
581        let start = stub.find("<lc_expand:").unwrap();
582        let end = stub[start..].find('>').unwrap() + start + 1;
583        let marker = &stub[start..end];
584
585        // ECHO + SPLICE: the model echoes the marker; the proxy splices the
586        // verbatim original (recovered from the local tee store) back inline.
587        let echo = serde_json::json!({
588            "messages": [
589                {"role": "user", "content": [{"type": "text", "text": "look again"}]},
590                {"role": "assistant", "content": format!("revisiting that output: {marker}")}
591            ]
592        });
593        let bytes = serde_json::to_vec(&echo).unwrap();
594        let (out, _o, _c) = compress_request_body(echo, bytes.len());
595        let spliced: Value = serde_json::from_slice(&out).unwrap();
596        let assistant = spliced["messages"][1]["content"].as_str().unwrap();
597        assert!(
598            assistant.contains("processed item 0007 ok")
599                && assistant.contains("processed item 0199 ok"),
600            "the verbatim original must be spliced back in full: {assistant}"
601        );
602        assert!(
603            !assistant.contains("<lc_expand:"),
604            "the marker must be consumed by the splice"
605        );
606    }
607
608    #[test]
609    fn inband_marker_less_turn_is_byte_identical_on_or_off() {
610        // Cache-safety (#493): enabling in-band must be a strict no-op on a turn
611        // with no marker — same bytes on the wire, so the provider cache prefix is
612        // never perturbed unless the model actually asked to expand. Uses a body
613        // with nothing to prune/compress, isolating the splice from stub emission.
614        let _iso = crate::core::data_dir::isolated_data_dir();
615        crate::test_env::remove_var("LEAN_CTX_PROXY_CCR_INBAND");
616        let body = serde_json::json!({
617            "messages": [
618                {"role": "user", "content": [{"type": "text", "text": "hello there"}]},
619                {"role": "assistant", "content": "hi — how can I help?"}
620            ]
621        });
622        let bytes = serde_json::to_vec(&body).unwrap();
623
624        crate::core::config::Config::update_global(|c| c.proxy.ccr_inband = Some(false)).unwrap();
625        let off = compress_request_body(body.clone(), bytes.len()).0;
626        crate::core::config::Config::update_global(|c| c.proxy.ccr_inband = Some(true)).unwrap();
627        let on = compress_request_body(body, bytes.len()).0;
628
629        assert_eq!(
630            off, on,
631            "a marker-less request must be byte-identical whether in-band is on or off"
632        );
633    }
634
635    /// Long, duplicate-rich natural-language prose that compresses cleanly.
636    fn big_prose() -> String {
637        let p = "You are a careful, senior software engineer. You always explain your \
638                 reasoning before making changes, you prefer small reviewable diffs, and \
639                 you never introduce mock data or placeholders into production code. ";
640        [p; 6].join("\n")
641    }
642
643    #[test]
644    fn system_prose_compressed_and_assistant_untouched() {
645        let _iso = crate::core::data_dir::isolated_data_dir();
646        crate::core::config::Config::update_global(|c| {
647            c.proxy.role_aggressiveness.system = Some(0.6);
648            c.proxy.role_aggressiveness.user = Some(0.6);
649        })
650        .unwrap();
651
652        let prose = big_prose();
653        let assistant_text = big_prose();
654        let body = serde_json::json!({
655            "model": "claude-opus-4-8",
656            "system": prose,
657            "messages": [
658                {"role": "user", "content": [{"type": "text", "text": prose}]},
659                {"role": "assistant", "content": assistant_text},
660            ]
661        });
662        let bytes = serde_json::to_vec(&body).unwrap();
663        let (out, _orig, _comp) = compress_request_body(body, bytes.len());
664        let parsed: Value = serde_json::from_slice(&out).unwrap();
665
666        assert!(
667            parsed["system"].as_str().unwrap().len() < prose.len(),
668            "system prose must be compressed when enabled"
669        );
670        assert_eq!(
671            parsed["messages"][1]["content"].as_str().unwrap(),
672            assistant_text,
673            "assistant turns must pass through verbatim (#710)"
674        );
675    }
676
677    #[test]
678    fn user_prose_compressed_only_in_frozen_region() {
679        let _iso = crate::core::data_dir::isolated_data_dir();
680        crate::core::config::Config::update_global(|c| {
681            c.proxy.role_aggressiveness.user = Some(0.7);
682        })
683        .unwrap();
684
685        let prose = big_prose();
686        // 30 messages → cache-aware boundary = ((30-8)/16)*16 = 16.
687        let mut messages = Vec::new();
688        for i in 0..30 {
689            let role = if i % 2 == 0 { "user" } else { "assistant" };
690            messages.push(serde_json::json!({
691                "role": role,
692                "content": [{"type": "text", "text": prose}]
693            }));
694        }
695        let body = serde_json::json!({ "messages": messages });
696        let bytes = serde_json::to_vec(&body).unwrap();
697        let (out, _o, _c) = compress_request_body(body, bytes.len());
698        let parsed: Value = serde_json::from_slice(&out).unwrap();
699
700        let frozen_user = parsed["messages"][0]["content"][0]["text"]
701            .as_str()
702            .unwrap();
703        assert!(
704            frozen_user.len() < prose.len(),
705            "user prose in the frozen region must be compressed"
706        );
707        assert_eq!(
708            parsed["messages"][1]["content"][0]["text"]
709                .as_str()
710                .unwrap(),
711            prose,
712            "assistant prose is never compressed"
713        );
714        let live_tail_user = parsed["messages"][28]["content"][0]["text"]
715            .as_str()
716            .unwrap();
717        assert_eq!(
718            live_tail_user, prose,
719            "user prose in the live tail (>= boundary) must be preserved for quality"
720        );
721    }
722
723    #[test]
724    fn client_cached_prefix_disables_system_prose() {
725        let _iso = crate::core::data_dir::isolated_data_dir();
726        crate::core::config::Config::update_global(|c| {
727            c.proxy.role_aggressiveness.system = Some(0.9);
728        })
729        .unwrap();
730
731        let prose = big_prose();
732        let body = serde_json::json!({
733            "system": prose,
734            "messages": [
735                {"role": "user", "content": [
736                    {"type": "text", "text": "hi", "cache_control": {"type": "ephemeral"}}
737                ]},
738                {"role": "assistant", "content": "ok"}
739            ]
740        });
741        let bytes = serde_json::to_vec(&body).unwrap();
742        let (out, _o, _c) = compress_request_body(body, bytes.len());
743        let parsed: Value = serde_json::from_slice(&out).unwrap();
744        assert_eq!(
745            parsed["system"].as_str().unwrap(),
746            prose,
747            "system must stay verbatim when the client caches a message prefix (#448)"
748        );
749    }
750
751    #[test]
752    fn prose_compression_is_deterministic() {
753        let _iso = crate::core::data_dir::isolated_data_dir();
754        crate::core::config::Config::update_global(|c| {
755            c.proxy.role_aggressiveness.system = Some(0.6);
756        })
757        .unwrap();
758        let prose = big_prose();
759        let mk = || serde_json::json!({"system": prose, "messages": [{"role": "user", "content": "hi"}]});
760        let (a, b) = (mk(), mk());
761        let la = serde_json::to_vec(&a).unwrap().len();
762        let lb = serde_json::to_vec(&b).unwrap().len();
763        assert_eq!(
764            compress_request_body(a, la).0,
765            compress_request_body(b, lb).0,
766            "prose compression must be byte-identical for identical input (#498)"
767        );
768    }
769
770    #[test]
771    fn bash_tool_result_still_compresses() {
772        let log = {
773            let mut s = String::from(
774                "$ git status\nOn branch main\nYour branch is up to date with 'origin/main'.\n\nChanges not staged for commit:\n  (use \"git add <file>...\" to update what will be committed)\n",
775            );
776            for i in 0..90 {
777                s.push_str(&format!("\tmodified:   src/module_{i}/file_{i}.rs\n"));
778            }
779            s.push_str("\nno changes added to commit (use \"git add\" and/or \"git commit -a\")\n");
780            s
781        };
782        let body = serde_json::json!({
783            "messages": [
784                {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "Bash", "input": {}}]},
785                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": log}]}
786            ]
787        });
788        let bytes = serde_json::to_vec(&body).unwrap();
789        let (_out, orig, comp) = compress_request_body(body, bytes.len());
790        assert!(comp < orig, "shell output must still be compressed");
791    }
792
793    #[test]
794    fn json_envelope_tool_result_is_compressed() {
795        let _iso = crate::core::data_dir::isolated_data_dir();
796        let log = long_git_status();
797        let expected = compress_tool_result(&log, Some("Bash"));
798        let envelope = serde_json::to_string(&serde_json::json!({
799            "content": [{"type": "text", "text": log}],
800            "isError": false,
801        }))
802        .unwrap();
803        let body = serde_json::json!({
804            "messages": [
805                {"role": "assistant", "content": [{
806                    "type": "tool_use",
807                    "id": "t1",
808                    "name": "Bash",
809                    "input": {}
810                }]},
811                {"role": "user", "content": [{
812                    "type": "tool_result",
813                    "tool_use_id": "t1",
814                    "content": envelope
815                }]}
816            ]
817        });
818        let bytes = serde_json::to_vec(&body).unwrap();
819        let (out, orig, comp) = compress_request_body(body, bytes.len());
820
821        assert!(comp < orig, "JSON envelope tool result should shrink");
822        let parsed: Value = serde_json::from_slice(&out).unwrap();
823        let content = parsed["messages"][1]["content"][0]["content"]
824            .as_str()
825            .unwrap();
826        let envelope: Value = serde_json::from_str(content).unwrap();
827        assert_eq!(envelope["content"][0]["text"].as_str().unwrap(), expected);
828    }
829
830    fn long_git_status() -> String {
831        let mut s = String::from(
832            "$ git status\nOn branch main\nYour branch is up to date with 'origin/main'.\n\nChanges not staged for commit:\n  (use \"git add <file>...\" to update what will be committed)\n",
833        );
834        for i in 0..80 {
835            s.push_str(&format!("\tmodified:   src/module_{i}/file_{i}.rs\n"));
836        }
837        s.push_str("\nno changes added to commit (use \"git add\" and/or \"git commit -a\")\n");
838        s
839    }
840
841    /// A client-cached message anchors the prefix; `system` precedes it, so the
842    /// cached prefix is `cached > 0` and system prose is normally protected.
843    /// System-prose verbatim-vs-rewritten is therefore a clean binary signal for
844    /// whether the #480 cold-prefix repack fired.
845    ///
846    /// `first_text` must be UNIQUE per test: it is `messages[0]`, which the
847    /// cold-prefix tracker hashes into the conversation key. A shared global
848    /// last-touch store has no test-clear hook (that would race with the unit
849    /// tests), so distinct keys are how parallel tests stay isolated.
850    fn cached_prefix_body(first_text: &str, prose: &str) -> (Vec<Value>, Value) {
851        let messages = vec![
852            serde_json::json!({"role": "user", "content": [
853                {"type": "text", "text": first_text, "cache_control": {"type": "ephemeral"}}
854            ]}),
855            serde_json::json!({"role": "assistant", "content": "ok"}),
856        ];
857        let body = serde_json::json!({ "system": prose, "messages": messages.clone() });
858        (messages, body)
859    }
860
861    #[test]
862    fn cold_prefix_repack_rewrites_protected_system_prose_when_enabled() {
863        let _iso = crate::core::data_dir::isolated_data_dir();
864        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
865        crate::core::config::Config::update_global(|c| {
866            c.proxy.role_aggressiveness.system = Some(0.9);
867            c.proxy.cold_prefix_repack = Some(true);
868        })
869        .unwrap();
870
871        // The prefix must clear the cacheable floor: with premium defaults the
872        // net-cost gate (#986, on by default) skips repacking a sub-1024-token
873        // prefix the provider could never cache. A real cold prefix worth
874        // re-seeding is large, so size the system prose accordingly.
875        let prose = big_prose().repeat(6);
876        let (messages, body) = cached_prefix_body("cold-repack-enabled-session", &prose);
877        // Predict cold: last touched 3h ago, well past the 5m default TTL × margin.
878        super::super::cold_prefix::test_seed_last_touch(&messages, 3 * 60 * 60);
879
880        let bytes = serde_json::to_vec(&body).unwrap();
881        let (out, _o, _c) = compress_request_body(body, bytes.len());
882        let parsed: Value = serde_json::from_slice(&out).unwrap();
883        assert!(
884            parsed["system"].as_str().unwrap().len() < prose.len(),
885            "a predicted-cold prefix must let the proxy repack the otherwise-protected system prose"
886        );
887    }
888
889    #[test]
890    fn cold_prefix_repack_skipped_for_subcacheable_prefix_by_default() {
891        // #986 premium default: cache_policy is on, so the net-cost gate skips a
892        // cold repack of a prefix below the provider's cacheable minimum —
893        // re-seeding it could never produce a cache the provider keeps. Repack is
894        // enabled and the prefix is cold, but it is too small (≈345 tokens) to
895        // cache, so the system prose must stay protected (unchanged).
896        let _iso = crate::core::data_dir::isolated_data_dir();
897        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
898        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_POLICY");
899        crate::core::config::Config::update_global(|c| {
900            c.proxy.role_aggressiveness.system = Some(0.9);
901            c.proxy.cold_prefix_repack = Some(true);
902        })
903        .unwrap();
904
905        let prose = big_prose();
906        let (messages, body) = cached_prefix_body("cold-repack-subcacheable-session", &prose);
907        super::super::cold_prefix::test_seed_last_touch(&messages, 3 * 60 * 60);
908
909        let bytes = serde_json::to_vec(&body).unwrap();
910        let (out, _o, _c) = compress_request_body(body, bytes.len());
911        let parsed: Value = serde_json::from_slice(&out).unwrap();
912        assert_eq!(
913            parsed["system"].as_str().unwrap(),
914            prose,
915            "the net-cost gate must skip repacking a sub-cacheable prefix (premium default)"
916        );
917    }
918
919    #[test]
920    fn cold_prefix_repack_off_by_default_keeps_prefix_protected() {
921        let _iso = crate::core::data_dir::isolated_data_dir();
922        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
923        crate::core::config::Config::update_global(|c| {
924            c.proxy.role_aggressiveness.system = Some(0.9);
925            c.proxy.cold_prefix_repack = Some(false);
926        })
927        .unwrap();
928
929        let prose = big_prose();
930        let (messages, body) = cached_prefix_body("cold-repack-disabled-session", &prose);
931        // Even with a huge idle gap, default-off must never touch the prefix.
932        super::super::cold_prefix::test_seed_last_touch(&messages, 24 * 60 * 60);
933
934        let bytes = serde_json::to_vec(&body).unwrap();
935        let (out, _o, _c) = compress_request_body(body, bytes.len());
936        let parsed: Value = serde_json::from_slice(&out).unwrap();
937        assert_eq!(
938            parsed["system"].as_str().unwrap(),
939            prose,
940            "with repack off the cached prefix stays byte-stable regardless of idle time (#448)"
941        );
942    }
943
944    #[test]
945    fn cold_prefix_repack_protects_warm_prefix_even_when_enabled() {
946        let _iso = crate::core::data_dir::isolated_data_dir();
947        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
948        crate::core::config::Config::update_global(|c| {
949            c.proxy.role_aggressiveness.system = Some(0.9);
950            c.proxy.cold_prefix_repack = Some(true);
951        })
952        .unwrap();
953
954        let prose = big_prose();
955        let (messages, body) = cached_prefix_body("cold-repack-warm-session", &prose);
956        // Warm: touched 1 minute ago → the prediction must keep protecting.
957        super::super::cold_prefix::test_seed_last_touch(&messages, 60);
958
959        let bytes = serde_json::to_vec(&body).unwrap();
960        let (out, _o, _c) = compress_request_body(body, bytes.len());
961        let parsed: Value = serde_json::from_slice(&out).unwrap();
962        assert_eq!(
963            parsed["system"].as_str().unwrap(),
964            prose,
965            "a warm prefix must stay protected even with repack enabled — only LARGE gaps trigger"
966        );
967    }
968
969    #[test]
970    fn cache_policy_attribution_is_measurement_only() {
971        // #986: enabling cache-economics records miss-attribution telemetry but
972        // must never change the bytes on the wire — the same request compressed
973        // with the policy off vs on is byte-identical (strictly cache-safe).
974        let _iso = crate::core::data_dir::isolated_data_dir();
975        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_POLICY");
976        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
977
978        let prose = big_prose();
979        let (_messages, body) = cached_prefix_body("cache-policy-measurement-session", &prose);
980        let bytes = serde_json::to_vec(&body).unwrap();
981
982        crate::core::config::Config::update_global(|c| {
983            c.proxy.cache_policy = Some(false);
984        })
985        .unwrap();
986        let (off, _o, _c) = compress_request_body(body.clone(), bytes.len());
987
988        crate::core::config::Config::update_global(|c| {
989            c.proxy.cache_policy = Some(true);
990        })
991        .unwrap();
992        let (on, _o, _c) = compress_request_body(body, bytes.len());
993
994        assert_eq!(
995            off, on,
996            "miss attribution is measurement-only: the wire bytes must not change"
997        );
998    }
999
1000    #[test]
1001    fn effort_control_dials_adaptive_thinking_only() {
1002        // #834 end-to-end: fill output_config.effort when the client already
1003        // asked for adaptive thinking, but never enable thinking otherwise.
1004        let _iso = crate::core::data_dir::isolated_data_dir();
1005        crate::test_env::remove_var("LEAN_CTX_PROXY_EFFORT");
1006        crate::core::config::Config::update_global(|c| {
1007            c.proxy.effort = Some("medium".into());
1008        })
1009        .unwrap();
1010
1011        let adaptive = serde_json::json!({
1012            "model": "claude-opus-4-8",
1013            "thinking": {"type": "adaptive"},
1014            "messages": [{"role": "user", "content": "hi"}]
1015        });
1016        let bytes = serde_json::to_vec(&adaptive).unwrap();
1017        let (out, _o, _c) = compress_request_body(adaptive, bytes.len());
1018        assert_eq!(
1019            serde_json::from_slice::<Value>(&out).unwrap()["output_config"]["effort"],
1020            "medium"
1021        );
1022
1023        // No thinking field → the proxy must not add output_config (no surprise
1024        // reasoning cost, no 400 risk).
1025        let plain = serde_json::json!({
1026            "model": "claude-opus-4-8",
1027            "messages": [{"role": "user", "content": "hi"}]
1028        });
1029        let bytes = serde_json::to_vec(&plain).unwrap();
1030        let (out, _o, _c) = compress_request_body(plain, bytes.len());
1031        assert!(
1032            serde_json::from_slice::<Value>(&out)
1033                .unwrap()
1034                .get("output_config")
1035                .is_none()
1036        );
1037    }
1038
1039    #[test]
1040    fn verbosity_steer_applies_to_treatment_skips_control() {
1041        // #895: the holdout control arm must be byte-unchanged (so its output is
1042        // the measurement baseline); the treatment arm gets the constant steer.
1043        let _iso = crate::core::data_dir::isolated_data_dir();
1044        crate::test_env::remove_var("LEAN_CTX_PROXY_VERBOSITY_STEER");
1045        crate::test_env::remove_var("LEAN_CTX_PROXY_OUTPUT_HOLDOUT");
1046        crate::test_env::remove_var("LEAN_CTX_PROXY_EFFORT");
1047
1048        let req = serde_json::json!({
1049            "model": "claude-opus-4-8",
1050            "messages": [{"role": "user", "content": "Summarize the design."}]
1051        });
1052        let bytes = serde_json::to_vec(&req).unwrap();
1053
1054        // Steer on, holdout = 0 → everyone Treatment → steered.
1055        crate::core::config::Config::update_global(|c| {
1056            c.proxy.verbosity_steer = Some(true);
1057            c.proxy.output_holdout = Some(0.0);
1058        })
1059        .unwrap();
1060        let (out, _o, _c) = compress_request_body(req.clone(), bytes.len());
1061        let v: Value = serde_json::from_slice(&out).unwrap();
1062        assert!(
1063            v["messages"][0]["content"]
1064                .as_str()
1065                .unwrap()
1066                .contains(crate::proxy::verbosity::STEER),
1067            "treatment arm must receive the verbosity steer"
1068        );
1069
1070        // Steer on, holdout = 1.0 → everyone Control → byte-unchanged, no steer.
1071        crate::core::config::Config::update_global(|c| {
1072            c.proxy.output_holdout = Some(1.0);
1073        })
1074        .unwrap();
1075        let (out2, _o, _c) = compress_request_body(req, bytes.len());
1076        let v2: Value = serde_json::from_slice(&out2).unwrap();
1077        assert!(
1078            !v2["messages"][0]["content"]
1079                .as_str()
1080                .unwrap()
1081                .contains(crate::proxy::verbosity::STEER),
1082            "control arm must NOT be steered (measurement baseline)"
1083        );
1084    }
1085
1086    /// A system prompt comfortably over Anthropic's minimum cacheable size, so
1087    /// the #939 breakpoint gate fires.
1088    fn cacheable_system() -> String {
1089        "You are a careful, senior software engineer who writes maintainable code. ".repeat(400)
1090    }
1091
1092    #[test]
1093    fn cache_breakpoint_injected_on_unanchored_system_when_opt_in() {
1094        // #939: opt-in on AND the client set no cache_control of its own → exactly
1095        // one ephemeral breakpoint lands on `system`, wrapping the verbatim text.
1096        let _iso = crate::core::data_dir::isolated_data_dir();
1097        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
1098        let body = serde_json::json!({
1099            "model": "claude-opus-4-8",
1100            "system": cacheable_system(),
1101            "messages": [{"role": "user", "content": "Refactor the parser."}]
1102        });
1103        let bytes = serde_json::to_vec(&body).unwrap();
1104
1105        crate::core::config::Config::update_global(|c| c.proxy.cache_breakpoint = Some(true))
1106            .unwrap();
1107        let (out, _o, _c) = compress_request_body(body, bytes.len());
1108        let v: Value = serde_json::from_slice(&out).unwrap();
1109
1110        assert_eq!(
1111            v["system"][0]["cache_control"]["type"], "ephemeral",
1112            "an unanchored system prompt must receive one ephemeral breakpoint"
1113        );
1114        assert!(
1115            v["system"][0]["text"]
1116                .as_str()
1117                .unwrap()
1118                .contains("senior software engineer"),
1119            "the system text must be preserved verbatim under the marker"
1120        );
1121    }
1122
1123    #[test]
1124    fn cache_breakpoint_off_by_default_is_byte_unchanged() {
1125        // Default off → the request must be byte-identical (no system reshape),
1126        // preserving the meter-only / cache-stable contract.
1127        let _iso = crate::core::data_dir::isolated_data_dir();
1128        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
1129        let body = serde_json::json!({
1130            "model": "claude-opus-4-8",
1131            "system": cacheable_system(),
1132            "messages": [{"role": "user", "content": "Refactor the parser."}]
1133        });
1134        let bytes = serde_json::to_vec(&body).unwrap();
1135        let (out, _o, _c) = compress_request_body(body, bytes.len());
1136        assert_eq!(
1137            out, bytes,
1138            "default-off must leave the request byte-identical"
1139        );
1140    }
1141
1142    #[test]
1143    fn cache_breakpoint_respects_client_anchor() {
1144        // #939 safety: a client cache_control on a message means `system` is part
1145        // of the already-cached prefix — never add a second, prefix-shifting anchor.
1146        let _iso = crate::core::data_dir::isolated_data_dir();
1147        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
1148        let body = serde_json::json!({
1149            "model": "claude-opus-4-8",
1150            "system": cacheable_system(),
1151            "messages": [{
1152                "role": "user",
1153                "content": [{
1154                    "type": "text",
1155                    "text": "hello",
1156                    "cache_control": {"type": "ephemeral"}
1157                }]
1158            }]
1159        });
1160        let bytes = serde_json::to_vec(&body).unwrap();
1161        crate::core::config::Config::update_global(|c| c.proxy.cache_breakpoint = Some(true))
1162            .unwrap();
1163        let (out, _o, _c) = compress_request_body(body, bytes.len());
1164        let v: Value = serde_json::from_slice(&out).unwrap();
1165        assert!(
1166            v["system"].is_string(),
1167            "with a client anchor present, system must be left untouched (no second breakpoint)"
1168        );
1169    }
1170
1171    #[test]
1172    fn cache_aligner_measures_without_mutating_body() {
1173        // #940: the volatile-field scan is telemetry-only — enabling it must leave
1174        // the request byte-identical (measurement, not a rewrite), even on a
1175        // volatile-field-rich system prompt.
1176        let _iso = crate::core::data_dir::isolated_data_dir();
1177        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_ALIGNER");
1178        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
1179        let body = serde_json::json!({
1180            "model": "claude-opus-4-8",
1181            "system": "Today is 2026-06-22. Session 550e8400-e29b-41d4-a716-446655440000.",
1182            "messages": [{"role": "user", "content": "Hello."}]
1183        });
1184        let bytes = serde_json::to_vec(&body).unwrap();
1185        crate::core::config::Config::update_global(|c| c.proxy.cache_aligner = Some(true)).unwrap();
1186        let (out, _o, _c) = compress_request_body(body, bytes.len());
1187        assert_eq!(
1188            out, bytes,
1189            "cache-aligner telemetry must never mutate the request body"
1190        );
1191    }
1192
1193    /// A cacheable-size system prompt that carries one volatile field (a date),
1194    /// so the #974 relocate has something to move and clears the size floor.
1195    fn cacheable_system_with_date() -> String {
1196        format!("Today is 2026-06-27. {}", cacheable_system())
1197    }
1198
1199    fn clear_relocate_env() {
1200        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_ALIGN_RELOCATE");
1201        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
1202        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_ALIGNER");
1203        crate::test_env::remove_var("LEAN_CTX_PROXY_OUTPUT_HOLDOUT");
1204    }
1205
1206    #[test]
1207    fn cache_align_relocate_moves_volatiles_to_tail_when_opt_in() {
1208        // #974: opt-in on, client anchored nothing → the date leaves the cacheable
1209        // prefix for an uncached tail block, and the stable block carries the one
1210        // ephemeral breakpoint.
1211        let _iso = crate::core::data_dir::isolated_data_dir();
1212        clear_relocate_env();
1213        let body = serde_json::json!({
1214            "model": "claude-opus-4-8",
1215            "system": cacheable_system_with_date(),
1216            "messages": [{"role": "user", "content": "Refactor the parser."}]
1217        });
1218        let bytes = serde_json::to_vec(&body).unwrap();
1219        crate::core::config::Config::update_global(|c| {
1220            c.proxy.cache_align_relocate = Some(true);
1221            c.proxy.output_holdout = Some(0.0);
1222        })
1223        .unwrap();
1224        let (out, _o, _c) = compress_request_body(body, bytes.len());
1225        let v: Value = serde_json::from_slice(&out).unwrap();
1226
1227        assert!(v["system"].is_array(), "system reshaped into a block array");
1228        assert_eq!(v["system"][0]["cache_control"]["type"], "ephemeral");
1229        assert!(
1230            !v["system"][0]["text"]
1231                .as_str()
1232                .unwrap()
1233                .contains("2026-06-27"),
1234            "the volatile date must leave the cacheable prefix"
1235        );
1236        assert!(
1237            v["system"][1].get("cache_control").is_none(),
1238            "the relocated tail block stays uncached"
1239        );
1240        assert!(
1241            v["system"][1]["text"]
1242                .as_str()
1243                .unwrap()
1244                .contains("2026-06-27"),
1245            "the date must be re-stated in the tail"
1246        );
1247    }
1248
1249    #[test]
1250    fn cache_align_relocate_off_by_default_is_byte_unchanged() {
1251        // Default off → byte-identical request, preserving the cache-stable
1252        // contract even with a volatile-rich system prompt.
1253        let _iso = crate::core::data_dir::isolated_data_dir();
1254        clear_relocate_env();
1255        let body = serde_json::json!({
1256            "model": "claude-opus-4-8",
1257            "system": cacheable_system_with_date(),
1258            "messages": [{"role": "user", "content": "Refactor the parser."}]
1259        });
1260        let bytes = serde_json::to_vec(&body).unwrap();
1261        let (out, _o, _c) = compress_request_body(body, bytes.len());
1262        assert_eq!(
1263            out, bytes,
1264            "default-off must leave the request byte-identical"
1265        );
1266    }
1267
1268    #[test]
1269    fn cache_align_relocate_skips_control_arm() {
1270        // #895 holdout: a control-arm conversation must be byte-unchanged so its
1271        // cache behaviour is the measurement baseline.
1272        let _iso = crate::core::data_dir::isolated_data_dir();
1273        clear_relocate_env();
1274        let body = serde_json::json!({
1275            "model": "claude-opus-4-8",
1276            "system": cacheable_system_with_date(),
1277            "messages": [{"role": "user", "content": "Refactor the parser."}]
1278        });
1279        let bytes = serde_json::to_vec(&body).unwrap();
1280        crate::core::config::Config::update_global(|c| {
1281            c.proxy.cache_align_relocate = Some(true);
1282            c.proxy.output_holdout = Some(1.0);
1283        })
1284        .unwrap();
1285        let (out, _o, _c) = compress_request_body(body, bytes.len());
1286        let v: Value = serde_json::from_slice(&out).unwrap();
1287        assert!(
1288            v["system"].is_string(),
1289            "control arm must not be relocated (measurement baseline)"
1290        );
1291    }
1292
1293    #[test]
1294    fn cache_align_relocate_respects_client_anchor() {
1295        // Safety: a client cache_control means `system` is part of the cached
1296        // prefix — never relocate it (that would shift the cached prefix, #448).
1297        let _iso = crate::core::data_dir::isolated_data_dir();
1298        clear_relocate_env();
1299        let body = serde_json::json!({
1300            "model": "claude-opus-4-8",
1301            "system": cacheable_system_with_date(),
1302            "messages": [{
1303                "role": "user",
1304                "content": [{
1305                    "type": "text",
1306                    "text": "hello",
1307                    "cache_control": {"type": "ephemeral"}
1308                }]
1309            }]
1310        });
1311        let bytes = serde_json::to_vec(&body).unwrap();
1312        crate::core::config::Config::update_global(|c| {
1313            c.proxy.cache_align_relocate = Some(true);
1314            c.proxy.output_holdout = Some(0.0);
1315        })
1316        .unwrap();
1317        let (out, _o, _c) = compress_request_body(body, bytes.len());
1318        let v: Value = serde_json::from_slice(&out).unwrap();
1319        assert!(
1320            v["system"].is_string(),
1321            "with a client anchor present, system must be left untouched"
1322        );
1323    }
1324
1325    #[test]
1326    fn cache_align_relocate_composes_with_breakpoint_to_one_anchor() {
1327        // Both opt-ins on: relocate adds the breakpoint to the stable block, so the
1328        // #939 injection sees an anchored prefix and stays a no-op — exactly one
1329        // breakpoint, on the stable block, with the volatile tail uncached.
1330        let _iso = crate::core::data_dir::isolated_data_dir();
1331        clear_relocate_env();
1332        let body = serde_json::json!({
1333            "model": "claude-opus-4-8",
1334            "system": cacheable_system_with_date(),
1335            "messages": [{"role": "user", "content": "Refactor the parser."}]
1336        });
1337        let bytes = serde_json::to_vec(&body).unwrap();
1338        crate::core::config::Config::update_global(|c| {
1339            c.proxy.cache_align_relocate = Some(true);
1340            c.proxy.cache_breakpoint = Some(true);
1341            c.proxy.output_holdout = Some(0.0);
1342        })
1343        .unwrap();
1344        let (out, _o, _c) = compress_request_body(body, bytes.len());
1345        let v: Value = serde_json::from_slice(&out).unwrap();
1346        let blocks = v["system"].as_array().expect("system is a block array");
1347        assert_eq!(blocks.len(), 2, "stable block + volatile tail");
1348        assert_eq!(blocks[0]["cache_control"]["type"], "ephemeral");
1349        assert!(
1350            blocks[1].get("cache_control").is_none(),
1351            "no second breakpoint — the tail stays uncached"
1352        );
1353    }
1354}