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

lean_ctx/proxy/
anthropic.rs
