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::compress_tool_result;
11use super::forward;
12use super::tool_kind::{self, ToolResultKind, should_protect};
13use super::{cache_safety, prose};
14use crate::core::config::{HistoryMode, ProseRole};
15
16pub async fn handler(
17    State(state): State<ProxyState>,
18    req: Request<Body>,
19) -> Result<Response, StatusCode> {
20    let upstream = state.anthropic_upstream();
21    forward::forward_request(
22        State(state),
23        req,
24        &upstream,
25        "/v1/messages",
26        compress_request_body,
27        "Anthropic",
28        &[],
29    )
30    .await
31}
32
33fn compress_request_body(parsed: Value, original_size: usize) -> (Vec<u8>, usize, usize) {
34    let mut doc = parsed;
35    let mut modified = false;
36
37    // Opt-in per-role prose aggressiveness (#710). Both default to `None`, in
38    // which case nothing below fires and the body is byte-for-byte unchanged.
39    let cfg = crate::core::config::Config::load();
40    let system_aggr = cfg.proxy.resolved_role_aggressiveness(ProseRole::System);
41    let user_aggr = cfg.proxy.resolved_role_aggressiveness(ProseRole::User);
42    let live_compress = cfg.proxy.live_compresses();
43    let mode = cfg.proxy.resolved_history_mode();
44    // #939: active prompt-cache breakpoint injection (opt-in, Anthropic-only).
45    // Resolved up front so the meter-only short-circuit below does not skip the
46    // one mutation this mode performs — its whole point is to add a cache anchor
47    // to an otherwise byte-passthrough request.
48    let inject_breakpoint = cfg.proxy.cache_breakpoint_enabled();
49    // #940: cache-aligner volatile-field telemetry (opt-in, measurement-only).
50    // Also resolved up front so a meter-only proxy still reaches the scan slot —
51    // it never mutates the body, it only records how much cache the system prompt
52    // leaks.
53    let align_volatile = cfg.proxy.cache_aligner_enabled();
54    // #895 Track B: output-savings holdout arm, from the pristine body (before any
55    // mutation below) so it matches the arm the response meter records. Control
56    // conversations skip output-shaping (effort + verbosity steer) but are still
57    // metered. Default holdout=0 → always Treatment (no behaviour change).
58    let arm = super::holdout::assign(
59        &super::holdout::anthropic_key(&doc),
60        cfg.proxy.output_holdout_fraction(),
61    );
62    // #493: in-band CCR expansion (opt-in). Splice any <lc_expand:HASH> the model
63    // echoed back into the verbatim original from the local tee store. A strict
64    // no-op when no marker is present (byte-identical body → cache-safe). Runs
65    // before the meter-only short-circuit so an explicit expand request is
66    // honored even when the proxy is otherwise byte-passthrough.
67    if cfg.proxy.ccr_inband_enabled() {
68        modified |= super::ccr::splice_inband_in_place(&mut doc);
69    }
70    // #834: cache-safe cross-provider effort control. Default off → no-op. The
71    // value is a constant, so it never perturbs the prompt-cache prefix; it only
72    // dials an *existing* adaptive thinking request (never enables thinking the
73    // client didn't ask for).
74    if arm == super::holdout::Arm::Treatment {
75        if let Some(effort) = cfg.proxy.resolved_effort() {
76            modified |= super::effort::apply_anthropic(&mut doc, effort);
77        }
78        // #895: cache-safe wire verbosity steer (constant suffix after the last
79        // cache_control breakpoint). Control arm skips it so the holdout measures
80        // its effect.
81        if cfg.proxy.verbosity_steer_enabled() {
82            modified |= super::verbosity::apply_anthropic(&mut doc);
83        }
84    }
85    // Meter-only (#481): live compression off, no history pruning, no prose
86    // rewriting → forward + usage metering still run, but the body is left
87    // unchanged so the provider prompt-cache prefix stays byte-stable. A pending
88    // in-band splice (`modified`) opts out: the body did change this turn.
89    if !live_compress
90        && mode == HistoryMode::Off
91        && system_aggr.is_none()
92        && user_aggr.is_none()
93        && !modified
94        && !inject_breakpoint
95        && !align_volatile
96    {
97        let out = serde_json::to_vec(&doc).unwrap_or_default();
98        return (out, original_size, original_size);
99    }
100    let mut prose_segments: u64 = 0;
101
102    // Length of the client's provider-cached message prefix. Needed both for
103    // cache-safe pruning below and to gate top-level system prose: if any
104    // message is client-cached, `system` (which precedes every message) is part
105    // of that cached prefix and must not be rewritten.
106    let cached = doc
107        .get("messages")
108        .and_then(|m| m.as_array())
109        .map_or(0, |m| super::history_prune::cached_prefix_len(m));
110
111    // #480: opt-in big-gap cold-prefix repack. When enabled AND the proxy can
112    // confidently predict (from idle time vs the provider cache TTL) that the
113    // client-cached prefix is already cold, override the normal "never touch the
114    // cached prefix" rule for THIS request and prune/compress the prefix too,
115    // re-seeding a leaner cache. Default-off; never fires without a measured idle
116    // gap past TTL × margin, so warm caches stay byte-stable (#448).
117    let repack = cfg.proxy.repacks_cold_prefix()
118        && doc
119            .get("messages")
120            .and_then(|m| m.as_array())
121            .is_some_and(|m| super::cold_prefix::repack_decision(m, cached));
122    // The prefix length the rewrites below must protect: the full cached prefix
123    // normally, or 0 when we are intentionally repacking the cold prefix.
124    let protect = if repack { 0 } else { cached };
125
126    // System prose: only when nothing is client-cached and the `system` field
127    // carries no `cache_control` of its own — otherwise it anchors the cache.
128    // A cold-prefix repack (`protect == 0` with `repack`) deliberately rewrites
129    // it to re-seed a leaner cache.
130    if let Some(a) = system_aggr
131        && protect == 0
132        && let Some(system) = doc.get_mut("system")
133        && (repack || !prose::value_has_cache_control(system))
134    {
135        let n = prose::compress_system_value(system, a);
136        if n > 0 {
137            prose_segments += u64::from(n);
138            modified = true;
139        }
140    }
141
142    if let Some(messages) = doc.get_mut("messages").and_then(|m| m.as_array_mut()) {
143        // Resolve tool-call id → tool name so file/source reads can be protected
144        // from lossy compression that would force the model to re-read mid-task.
145        let tool_names = tool_kind::anthropic_tool_names(messages);
146
147        // Prune at a frozen, cache-aware boundary by default: Anthropic's
148        // prompt cache matches exact prefixes, so the boundary must not move
149        // every turn (see `history_prune::prune_boundary`). `mode` resolved above.
150        let boundary = super::history_prune::prune_boundary(mode, messages.len());
151        // Never rewrite content the client has marked with `cache_control`:
152        // pruning inside the already-cached prefix invalidates Anthropic's
153        // prompt cache from the first changed message (#448). Pruning therefore
154        // starts after the last breakpoint; with no breakpoint this is 0, i.e.
155        // the previous behaviour.
156        modified |=
157            super::history_prune::prune_history_range(messages, protect, boundary, &tool_names);
158
159        for msg in messages.iter_mut() {
160            let role = msg.get("role").and_then(|r| r.as_str()).unwrap_or("");
161            if role != "user" {
162                continue;
163            }
164
165            if let Some(content) = msg.get_mut("content").and_then(|c| c.as_array_mut()) {
166                for block in content.iter_mut() {
167                    if block.get("type").and_then(|t| t.as_str()) != Some("tool_result") {
168                        continue;
169                    }
170
171                    let name = block
172                        .get("tool_use_id")
173                        .and_then(|v| v.as_str())
174                        .and_then(|id| tool_names.get(id))
175                        .map(String::as_str);
176                    let kind = name.map_or(ToolResultKind::Other, tool_kind::classify_tool_name);
177
178                    // #481: skip live compression when globally off or when the
179                    // originating tool is on the exclusion list (Serena default).
180                    let excluded =
181                        name.is_some_and(|n| cfg.proxy.is_tool_live_compress_excluded(n));
182                    if live_compress
183                        && !excluded
184                        && let Some(inner_content) = block.get_mut("content")
185                    {
186                        modified |= compress_content_field(inner_content, name, kind);
187                    }
188                }
189            }
190        }
191
192        // Frozen-region user prose: free-text `text` blocks of user turns in
193        // `[cached, boundary)`. Cache-safe by construction — the cached prefix
194        // and the live tail (`>= boundary`) are both left intact, and the
195        // rewrite is content-deterministic so the prefix stays byte-stable.
196        if let Some(a) = user_aggr {
197            let end = boundary.min(messages.len());
198            let start = protect.min(end);
199            for msg in &mut messages[start..end] {
200                if msg.get("role").and_then(|r| r.as_str()) == Some("user")
201                    && let Some(content) = msg.get_mut("content").and_then(|c| c.as_array_mut())
202                {
203                    prose_segments += u64::from(prose::compress_text_blocks(content, a));
204                }
205            }
206        }
207    }
208
209    if prose_segments > 0 {
210        modified = true;
211    }
212    // #940: cache-aligner telemetry. On an unanchored system prompt (the prefix a
213    // provider would cache), count the volatile fields that would bust that cache
214    // turn-to-turn. Pure measurement — runs before any breakpoint injection and
215    // never mutates the body — so it is strictly cache-safe. Skipped once the
216    // client has anchored the prefix itself.
217    if align_volatile
218        && cached == 0
219        && let Some(system) = doc.get("system")
220        && !prose::value_has_cache_control(system)
221        && let Some(text) = super::cache_aligner::system_text(system)
222    {
223        let scan = super::cache_aligner::scan_volatile(&text);
224        cache_safety::record_volatile_system(scan.fields as u64);
225    }
226    // #939: active prompt-cache breakpoint injection (Anthropic-only). When the
227    // client anchored no prefix of its own — no message `cache_control` (`cached
228    // == 0`) and no breakpoint already on `system` — add one ephemeral breakpoint
229    // to `system` so the large, stable system prompt bills later turns at the
230    // cached rate (the win a raw API client leaves on the table). Runs after every
231    // frozen-region rewrite so the marker anchors the final system bytes and the
232    // prefix it creates stays byte-stable across turns (#498). Counted on its own
233    // gauge — a pure win, never against the cache-safe ratio.
234    if inject_breakpoint
235        && cached == 0
236        && doc
237            .get("system")
238            .is_some_and(|s| !prose::value_has_cache_control(s))
239        && super::cache_breakpoint::inject_anthropic_system(&mut doc)
240    {
241        modified = true;
242        cache_safety::record_breakpoint_injected();
243    }
244    // A deliberate cold-prefix repack (#480) is the one sanctioned exception to
245    // the frozen-window rule; count it on its own gauge so it never dilutes the
246    // cache-safe ratio (which exists to catch *accidental* #448 regressions).
247    // Every other rewrite lands strictly inside the cache-safe frozen window.
248    if repack {
249        cache_safety::record_cold_repack();
250    }
251    cache_safety::record(prose_segments, true);
252
253    let out = serde_json::to_vec(&doc).unwrap_or_default();
254    let compressed_size = if modified { out.len() } else { original_size };
255    (out, original_size, compressed_size)
256}
257
258/// Compresses a tool_result `content` field unless it is a protected file/source
259/// read, which must reach the model intact (it is what gets edited).
260fn compress_content_field(
261    content: &mut Value,
262    tool_name: Option<&str>,
263    kind: ToolResultKind,
264) -> bool {
265    match content {
266        Value::String(s) => {
267            if should_protect(kind, s) {
268                return false;
269            }
270            let compressed = compress_tool_result(s, tool_name);
271            if compressed.len() < s.len() {
272                *s = compressed;
273                return true;
274            }
275            false
276        }
277        Value::Array(arr) => {
278            let mut modified = false;
279            for item in arr.iter_mut() {
280                if item.get("type").and_then(|t| t.as_str()) == Some("text")
281                    && let Some(text) = item
282                        .get_mut("text")
283                        .and_then(|t| t.as_str().map(String::from))
284                {
285                    if should_protect(kind, &text) {
286                        continue;
287                    }
288                    let compressed = compress_tool_result(&text, tool_name);
289                    if compressed.len() < text.len() {
290                        item["text"] = Value::String(compressed);
291                        modified = true;
292                    }
293                }
294            }
295            modified
296        }
297        _ => false,
298    }
299}
300
301#[cfg(test)]
302mod tests {
303    use super::*;
304
305    fn source_file_body() -> Vec<u8> {
306        let code = (0..60)
307            .map(|i| format!("    let binding_{i} = compute_value_{i}(context, options);"))
308            .collect::<Vec<_>>()
309            .join("\n");
310        let body = serde_json::json!({
311            "model": "claude-opus-4-8",
312            "messages": [
313                {
314                    "role": "assistant",
315                    "content": [{"type": "tool_use", "id": "toolu_1", "name": "Read", "input": {"file_path": "src/app.rs"}}]
316                },
317                {
318                    "role": "user",
319                    "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": code}]
320                }
321            ]
322        });
323        serde_json::to_vec(&body).unwrap()
324    }
325
326    #[test]
327    fn read_tool_result_is_never_truncated() {
328        let bytes = source_file_body();
329        let body: Value = serde_json::from_slice(&bytes).unwrap();
330        let (out, _orig, _comp) = compress_request_body(body, bytes.len());
331        let parsed: Value = serde_json::from_slice(&out).unwrap();
332        let content = parsed["messages"][1]["content"][0]["content"]
333            .as_str()
334            .unwrap();
335        assert!(
336            content.contains("binding_59"),
337            "the full source body must survive — refactors need it intact"
338        );
339        assert!(!content.contains("lines omitted"));
340    }
341
342    fn forge_log_body(tool_name: &str) -> Value {
343        // Generic, highly-repetitive log with no `$ cmd` hint, so routing falls
344        // back to the tool name (exercising the foreign-tool classification)
345        // and the generic compressor (not a command-specific pattern).
346        let mut log = String::new();
347        for i in 0..90 {
348            log.push_str(&format!(
349                "INFO  processing item {i}: ok, latency={i}ms, queue depth normal, retries 0\n"
350            ));
351        }
352        serde_json::json!({
353            "messages": [
354                {"role": "assistant", "content": [{"type": "tool_use", "id": "f1", "name": tool_name, "input": {}}]},
355                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "f1", "content": log}]}
356            ]
357        })
358    }
359
360    #[test]
361    fn forge_shell_tool_result_compresses() {
362        // A vendor-prefixed foreign shell tool reaches the proxy; its log output
363        // must still be compressed (rtk/ctx_* never see another server's tools).
364        let body = forge_log_body("forge_shell");
365        let bytes = serde_json::to_vec(&body).unwrap();
366        let (_out, orig, comp) = compress_request_body(body, bytes.len());
367        assert!(comp < orig, "foreign shell output must be compressed");
368    }
369
370    #[test]
371    fn foreign_read_tool_protects_source() {
372        // `forge_read` is classified FileRead via the segment fallback, so the
373        // source body must reach the model intact (it is what gets edited).
374        let code = (0..60)
375            .map(|i| format!("    let binding_{i} = compute_value_{i}(context, options);"))
376            .collect::<Vec<_>>()
377            .join("\n");
378        let body = serde_json::json!({
379            "messages": [
380                {"role": "assistant", "content": [{"type": "tool_use", "id": "r1", "name": "forge_read", "input": {"path": "src/app.rs"}}]},
381                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "r1", "content": code}]}
382            ]
383        });
384        let bytes = serde_json::to_vec(&body).unwrap();
385        let (out, _orig, _comp) = compress_request_body(body, bytes.len());
386        let parsed: Value = serde_json::from_slice(&out).unwrap();
387        let content = parsed["messages"][1]["content"][0]["content"]
388            .as_str()
389            .unwrap();
390        assert!(
391            content.contains("binding_59"),
392            "source body must survive intact"
393        );
394    }
395
396    #[test]
397    fn compress_request_body_is_deterministic() {
398        // tee path depends on the data dir; serialize env access so a parallel
399        // test never swaps LEAN_CTX_DATA_DIR between the two compressions.
400        let _lock = crate::core::data_dir::test_env_lock();
401        // #498: the proxy rewrite must be a pure function of the body so the
402        // provider prompt-cache prefix stays byte-identical across turns.
403        let bytes = serde_json::to_vec(&forge_log_body("Bash")).unwrap();
404        let a = compress_request_body(serde_json::from_slice(&bytes).unwrap(), bytes.len()).0;
405        let b = compress_request_body(serde_json::from_slice(&bytes).unwrap(), bytes.len()).0;
406        assert_eq!(a, b, "identical input must yield byte-identical output");
407    }
408
409    /// A large, highly-compressible foreign log so the live path tees + stubs it.
410    fn big_log() -> String {
411        (0..200)
412            .map(|i| format!("[info] processed item {i:04} ok, latency {i}ms, queue normal"))
413            .collect::<Vec<_>>()
414            .join("\n")
415    }
416
417    #[test]
418    fn inband_ccr_emit_echo_splice_round_trip() {
419        // Full #493 cycle through the real Anthropic request path: a lossy stub
420        // emits an <lc_expand:HASH> marker, the model echoes it, and the proxy
421        // splices the verbatim original back inline on the next request.
422        let _iso = crate::core::data_dir::isolated_data_dir();
423        crate::test_env::remove_var("LEAN_CTX_PROXY_CCR_INBAND");
424        crate::core::config::Config::update_global(|c| {
425            c.proxy.ccr_inband = Some(true);
426        })
427        .unwrap();
428
429        // EMIT: live-compress a foreign tool_result → recovery stub with a marker.
430        let log = big_log();
431        let emit = serde_json::json!({
432            "messages": [
433                {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "bash", "input": {}}]},
434                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": log}]}
435            ]
436        });
437        let bytes = serde_json::to_vec(&emit).unwrap();
438        let (out, _o, _c) = compress_request_body(emit, bytes.len());
439        let emitted: Value = serde_json::from_slice(&out).unwrap();
440        let stub = emitted["messages"][1]["content"][0]["content"]
441            .as_str()
442            .unwrap();
443        assert!(
444            stub.contains("<lc_expand:"),
445            "in-band stub must advertise an echo-able marker: {stub}"
446        );
447        assert!(
448            !stub.contains("/tee/proxy_"),
449            "in-band stub must not leak the unreachable local tee path: {stub}"
450        );
451
452        // The marker the model would copy into its next turn.
453        let start = stub.find("<lc_expand:").unwrap();
454        let end = stub[start..].find('>').unwrap() + start + 1;
455        let marker = &stub[start..end];
456
457        // ECHO + SPLICE: the model echoes the marker; the proxy splices the
458        // verbatim original (recovered from the local tee store) back inline.
459        let echo = serde_json::json!({
460            "messages": [
461                {"role": "user", "content": [{"type": "text", "text": "look again"}]},
462                {"role": "assistant", "content": format!("revisiting that output: {marker}")}
463            ]
464        });
465        let bytes = serde_json::to_vec(&echo).unwrap();
466        let (out, _o, _c) = compress_request_body(echo, bytes.len());
467        let spliced: Value = serde_json::from_slice(&out).unwrap();
468        let assistant = spliced["messages"][1]["content"].as_str().unwrap();
469        assert!(
470            assistant.contains("processed item 0007 ok")
471                && assistant.contains("processed item 0199 ok"),
472            "the verbatim original must be spliced back in full: {assistant}"
473        );
474        assert!(
475            !assistant.contains("<lc_expand:"),
476            "the marker must be consumed by the splice"
477        );
478    }
479
480    #[test]
481    fn inband_marker_less_turn_is_byte_identical_on_or_off() {
482        // Cache-safety (#493): enabling in-band must be a strict no-op on a turn
483        // with no marker — same bytes on the wire, so the provider cache prefix is
484        // never perturbed unless the model actually asked to expand. Uses a body
485        // with nothing to prune/compress, isolating the splice from stub emission.
486        let _iso = crate::core::data_dir::isolated_data_dir();
487        crate::test_env::remove_var("LEAN_CTX_PROXY_CCR_INBAND");
488        let body = serde_json::json!({
489            "messages": [
490                {"role": "user", "content": [{"type": "text", "text": "hello there"}]},
491                {"role": "assistant", "content": "hi — how can I help?"}
492            ]
493        });
494        let bytes = serde_json::to_vec(&body).unwrap();
495
496        crate::core::config::Config::update_global(|c| c.proxy.ccr_inband = Some(false)).unwrap();
497        let off = compress_request_body(body.clone(), bytes.len()).0;
498        crate::core::config::Config::update_global(|c| c.proxy.ccr_inband = Some(true)).unwrap();
499        let on = compress_request_body(body, bytes.len()).0;
500
501        assert_eq!(
502            off, on,
503            "a marker-less request must be byte-identical whether in-band is on or off"
504        );
505    }
506
507    /// Long, duplicate-rich natural-language prose that compresses cleanly.
508    fn big_prose() -> String {
509        let p = "You are a careful, senior software engineer. You always explain your \
510                 reasoning before making changes, you prefer small reviewable diffs, and \
511                 you never introduce mock data or placeholders into production code. ";
512        [p; 6].join("\n")
513    }
514
515    #[test]
516    fn system_prose_compressed_and_assistant_untouched() {
517        let _iso = crate::core::data_dir::isolated_data_dir();
518        crate::core::config::Config::update_global(|c| {
519            c.proxy.role_aggressiveness.system = Some(0.6);
520            c.proxy.role_aggressiveness.user = Some(0.6);
521        })
522        .unwrap();
523
524        let prose = big_prose();
525        let assistant_text = big_prose();
526        let body = serde_json::json!({
527            "model": "claude-opus-4-8",
528            "system": prose,
529            "messages": [
530                {"role": "user", "content": [{"type": "text", "text": prose}]},
531                {"role": "assistant", "content": assistant_text},
532            ]
533        });
534        let bytes = serde_json::to_vec(&body).unwrap();
535        let (out, _orig, _comp) = compress_request_body(body, bytes.len());
536        let parsed: Value = serde_json::from_slice(&out).unwrap();
537
538        assert!(
539            parsed["system"].as_str().unwrap().len() < prose.len(),
540            "system prose must be compressed when enabled"
541        );
542        assert_eq!(
543            parsed["messages"][1]["content"].as_str().unwrap(),
544            assistant_text,
545            "assistant turns must pass through verbatim (#710)"
546        );
547    }
548
549    #[test]
550    fn user_prose_compressed_only_in_frozen_region() {
551        let _iso = crate::core::data_dir::isolated_data_dir();
552        crate::core::config::Config::update_global(|c| {
553            c.proxy.role_aggressiveness.user = Some(0.7);
554        })
555        .unwrap();
556
557        let prose = big_prose();
558        // 30 messages → cache-aware boundary = ((30-8)/16)*16 = 16.
559        let mut messages = Vec::new();
560        for i in 0..30 {
561            let role = if i % 2 == 0 { "user" } else { "assistant" };
562            messages.push(serde_json::json!({
563                "role": role,
564                "content": [{"type": "text", "text": prose}]
565            }));
566        }
567        let body = serde_json::json!({ "messages": messages });
568        let bytes = serde_json::to_vec(&body).unwrap();
569        let (out, _o, _c) = compress_request_body(body, bytes.len());
570        let parsed: Value = serde_json::from_slice(&out).unwrap();
571
572        let frozen_user = parsed["messages"][0]["content"][0]["text"]
573            .as_str()
574            .unwrap();
575        assert!(
576            frozen_user.len() < prose.len(),
577            "user prose in the frozen region must be compressed"
578        );
579        assert_eq!(
580            parsed["messages"][1]["content"][0]["text"]
581                .as_str()
582                .unwrap(),
583            prose,
584            "assistant prose is never compressed"
585        );
586        let live_tail_user = parsed["messages"][28]["content"][0]["text"]
587            .as_str()
588            .unwrap();
589        assert_eq!(
590            live_tail_user, prose,
591            "user prose in the live tail (>= boundary) must be preserved for quality"
592        );
593    }
594
595    #[test]
596    fn client_cached_prefix_disables_system_prose() {
597        let _iso = crate::core::data_dir::isolated_data_dir();
598        crate::core::config::Config::update_global(|c| {
599            c.proxy.role_aggressiveness.system = Some(0.9);
600        })
601        .unwrap();
602
603        let prose = big_prose();
604        let body = serde_json::json!({
605            "system": prose,
606            "messages": [
607                {"role": "user", "content": [
608                    {"type": "text", "text": "hi", "cache_control": {"type": "ephemeral"}}
609                ]},
610                {"role": "assistant", "content": "ok"}
611            ]
612        });
613        let bytes = serde_json::to_vec(&body).unwrap();
614        let (out, _o, _c) = compress_request_body(body, bytes.len());
615        let parsed: Value = serde_json::from_slice(&out).unwrap();
616        assert_eq!(
617            parsed["system"].as_str().unwrap(),
618            prose,
619            "system must stay verbatim when the client caches a message prefix (#448)"
620        );
621    }
622
623    #[test]
624    fn prose_compression_is_deterministic() {
625        let _iso = crate::core::data_dir::isolated_data_dir();
626        crate::core::config::Config::update_global(|c| {
627            c.proxy.role_aggressiveness.system = Some(0.6);
628        })
629        .unwrap();
630        let prose = big_prose();
631        let mk = || serde_json::json!({"system": prose, "messages": [{"role": "user", "content": "hi"}]});
632        let (a, b) = (mk(), mk());
633        let la = serde_json::to_vec(&a).unwrap().len();
634        let lb = serde_json::to_vec(&b).unwrap().len();
635        assert_eq!(
636            compress_request_body(a, la).0,
637            compress_request_body(b, lb).0,
638            "prose compression must be byte-identical for identical input (#498)"
639        );
640    }
641
642    #[test]
643    fn bash_tool_result_still_compresses() {
644        let log = {
645            let mut s = String::from(
646                "$ 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",
647            );
648            for i in 0..90 {
649                s.push_str(&format!("\tmodified:   src/module_{i}/file_{i}.rs\n"));
650            }
651            s.push_str("\nno changes added to commit (use \"git add\" and/or \"git commit -a\")\n");
652            s
653        };
654        let body = serde_json::json!({
655            "messages": [
656                {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "Bash", "input": {}}]},
657                {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": log}]}
658            ]
659        });
660        let bytes = serde_json::to_vec(&body).unwrap();
661        let (_out, orig, comp) = compress_request_body(body, bytes.len());
662        assert!(comp < orig, "shell output must still be compressed");
663    }
664
665    /// A client-cached message anchors the prefix; `system` precedes it, so the
666    /// cached prefix is `cached > 0` and system prose is normally protected.
667    /// System-prose verbatim-vs-rewritten is therefore a clean binary signal for
668    /// whether the #480 cold-prefix repack fired.
669    ///
670    /// `first_text` must be UNIQUE per test: it is `messages[0]`, which the
671    /// cold-prefix tracker hashes into the conversation key. A shared global
672    /// last-touch store has no test-clear hook (that would race with the unit
673    /// tests), so distinct keys are how parallel tests stay isolated.
674    fn cached_prefix_body(first_text: &str, prose: &str) -> (Vec<Value>, Value) {
675        let messages = vec![
676            serde_json::json!({"role": "user", "content": [
677                {"type": "text", "text": first_text, "cache_control": {"type": "ephemeral"}}
678            ]}),
679            serde_json::json!({"role": "assistant", "content": "ok"}),
680        ];
681        let body = serde_json::json!({ "system": prose, "messages": messages.clone() });
682        (messages, body)
683    }
684
685    #[test]
686    fn cold_prefix_repack_rewrites_protected_system_prose_when_enabled() {
687        let _iso = crate::core::data_dir::isolated_data_dir();
688        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
689        crate::core::config::Config::update_global(|c| {
690            c.proxy.role_aggressiveness.system = Some(0.9);
691            c.proxy.cold_prefix_repack = Some(true);
692        })
693        .unwrap();
694
695        let prose = big_prose();
696        let (messages, body) = cached_prefix_body("cold-repack-enabled-session", &prose);
697        // Predict cold: last touched 3h ago, well past the 5m default TTL × margin.
698        super::super::cold_prefix::test_seed_last_touch(&messages, 3 * 60 * 60);
699
700        let bytes = serde_json::to_vec(&body).unwrap();
701        let (out, _o, _c) = compress_request_body(body, bytes.len());
702        let parsed: Value = serde_json::from_slice(&out).unwrap();
703        assert!(
704            parsed["system"].as_str().unwrap().len() < prose.len(),
705            "a predicted-cold prefix must let the proxy repack the otherwise-protected system prose"
706        );
707    }
708
709    #[test]
710    fn cold_prefix_repack_off_by_default_keeps_prefix_protected() {
711        let _iso = crate::core::data_dir::isolated_data_dir();
712        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
713        crate::core::config::Config::update_global(|c| {
714            c.proxy.role_aggressiveness.system = Some(0.9);
715            c.proxy.cold_prefix_repack = Some(false);
716        })
717        .unwrap();
718
719        let prose = big_prose();
720        let (messages, body) = cached_prefix_body("cold-repack-disabled-session", &prose);
721        // Even with a huge idle gap, default-off must never touch the prefix.
722        super::super::cold_prefix::test_seed_last_touch(&messages, 24 * 60 * 60);
723
724        let bytes = serde_json::to_vec(&body).unwrap();
725        let (out, _o, _c) = compress_request_body(body, bytes.len());
726        let parsed: Value = serde_json::from_slice(&out).unwrap();
727        assert_eq!(
728            parsed["system"].as_str().unwrap(),
729            prose,
730            "with repack off the cached prefix stays byte-stable regardless of idle time (#448)"
731        );
732    }
733
734    #[test]
735    fn cold_prefix_repack_protects_warm_prefix_even_when_enabled() {
736        let _iso = crate::core::data_dir::isolated_data_dir();
737        crate::test_env::remove_var("LEAN_CTX_PROXY_COLD_PREFIX_REPACK");
738        crate::core::config::Config::update_global(|c| {
739            c.proxy.role_aggressiveness.system = Some(0.9);
740            c.proxy.cold_prefix_repack = Some(true);
741        })
742        .unwrap();
743
744        let prose = big_prose();
745        let (messages, body) = cached_prefix_body("cold-repack-warm-session", &prose);
746        // Warm: touched 1 minute ago → the prediction must keep protecting.
747        super::super::cold_prefix::test_seed_last_touch(&messages, 60);
748
749        let bytes = serde_json::to_vec(&body).unwrap();
750        let (out, _o, _c) = compress_request_body(body, bytes.len());
751        let parsed: Value = serde_json::from_slice(&out).unwrap();
752        assert_eq!(
753            parsed["system"].as_str().unwrap(),
754            prose,
755            "a warm prefix must stay protected even with repack enabled — only LARGE gaps trigger"
756        );
757    }
758
759    #[test]
760    fn effort_control_dials_adaptive_thinking_only() {
761        // #834 end-to-end: fill output_config.effort when the client already
762        // asked for adaptive thinking, but never enable thinking otherwise.
763        let _iso = crate::core::data_dir::isolated_data_dir();
764        crate::test_env::remove_var("LEAN_CTX_PROXY_EFFORT");
765        crate::core::config::Config::update_global(|c| {
766            c.proxy.effort = Some("medium".into());
767        })
768        .unwrap();
769
770        let adaptive = serde_json::json!({
771            "model": "claude-opus-4-8",
772            "thinking": {"type": "adaptive"},
773            "messages": [{"role": "user", "content": "hi"}]
774        });
775        let bytes = serde_json::to_vec(&adaptive).unwrap();
776        let (out, _o, _c) = compress_request_body(adaptive, bytes.len());
777        assert_eq!(
778            serde_json::from_slice::<Value>(&out).unwrap()["output_config"]["effort"],
779            "medium"
780        );
781
782        // No thinking field → the proxy must not add output_config (no surprise
783        // reasoning cost, no 400 risk).
784        let plain = serde_json::json!({
785            "model": "claude-opus-4-8",
786            "messages": [{"role": "user", "content": "hi"}]
787        });
788        let bytes = serde_json::to_vec(&plain).unwrap();
789        let (out, _o, _c) = compress_request_body(plain, bytes.len());
790        assert!(
791            serde_json::from_slice::<Value>(&out)
792                .unwrap()
793                .get("output_config")
794                .is_none()
795        );
796    }
797
798    #[test]
799    fn verbosity_steer_applies_to_treatment_skips_control() {
800        // #895: the holdout control arm must be byte-unchanged (so its output is
801        // the measurement baseline); the treatment arm gets the constant steer.
802        let _iso = crate::core::data_dir::isolated_data_dir();
803        crate::test_env::remove_var("LEAN_CTX_PROXY_VERBOSITY_STEER");
804        crate::test_env::remove_var("LEAN_CTX_PROXY_OUTPUT_HOLDOUT");
805        crate::test_env::remove_var("LEAN_CTX_PROXY_EFFORT");
806
807        let req = serde_json::json!({
808            "model": "claude-opus-4-8",
809            "messages": [{"role": "user", "content": "Summarize the design."}]
810        });
811        let bytes = serde_json::to_vec(&req).unwrap();
812
813        // Steer on, holdout = 0 → everyone Treatment → steered.
814        crate::core::config::Config::update_global(|c| {
815            c.proxy.verbosity_steer = Some(true);
816            c.proxy.output_holdout = Some(0.0);
817        })
818        .unwrap();
819        let (out, _o, _c) = compress_request_body(req.clone(), bytes.len());
820        let v: Value = serde_json::from_slice(&out).unwrap();
821        assert!(
822            v["messages"][0]["content"]
823                .as_str()
824                .unwrap()
825                .contains(crate::proxy::verbosity::STEER),
826            "treatment arm must receive the verbosity steer"
827        );
828
829        // Steer on, holdout = 1.0 → everyone Control → byte-unchanged, no steer.
830        crate::core::config::Config::update_global(|c| {
831            c.proxy.output_holdout = Some(1.0);
832        })
833        .unwrap();
834        let (out2, _o, _c) = compress_request_body(req, bytes.len());
835        let v2: Value = serde_json::from_slice(&out2).unwrap();
836        assert!(
837            !v2["messages"][0]["content"]
838                .as_str()
839                .unwrap()
840                .contains(crate::proxy::verbosity::STEER),
841            "control arm must NOT be steered (measurement baseline)"
842        );
843    }
844
845    /// A system prompt comfortably over Anthropic's minimum cacheable size, so
846    /// the #939 breakpoint gate fires.
847    fn cacheable_system() -> String {
848        "You are a careful, senior software engineer who writes maintainable code. ".repeat(400)
849    }
850
851    #[test]
852    fn cache_breakpoint_injected_on_unanchored_system_when_opt_in() {
853        // #939: opt-in on AND the client set no cache_control of its own → exactly
854        // one ephemeral breakpoint lands on `system`, wrapping the verbatim text.
855        let _iso = crate::core::data_dir::isolated_data_dir();
856        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
857        let body = serde_json::json!({
858            "model": "claude-opus-4-8",
859            "system": cacheable_system(),
860            "messages": [{"role": "user", "content": "Refactor the parser."}]
861        });
862        let bytes = serde_json::to_vec(&body).unwrap();
863
864        crate::core::config::Config::update_global(|c| c.proxy.cache_breakpoint = Some(true))
865            .unwrap();
866        let (out, _o, _c) = compress_request_body(body, bytes.len());
867        let v: Value = serde_json::from_slice(&out).unwrap();
868
869        assert_eq!(
870            v["system"][0]["cache_control"]["type"], "ephemeral",
871            "an unanchored system prompt must receive one ephemeral breakpoint"
872        );
873        assert!(
874            v["system"][0]["text"]
875                .as_str()
876                .unwrap()
877                .contains("senior software engineer"),
878            "the system text must be preserved verbatim under the marker"
879        );
880    }
881
882    #[test]
883    fn cache_breakpoint_off_by_default_is_byte_unchanged() {
884        // Default off → the request must be byte-identical (no system reshape),
885        // preserving the meter-only / cache-stable contract.
886        let _iso = crate::core::data_dir::isolated_data_dir();
887        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
888        let body = serde_json::json!({
889            "model": "claude-opus-4-8",
890            "system": cacheable_system(),
891            "messages": [{"role": "user", "content": "Refactor the parser."}]
892        });
893        let bytes = serde_json::to_vec(&body).unwrap();
894        let (out, _o, _c) = compress_request_body(body, bytes.len());
895        assert_eq!(
896            out, bytes,
897            "default-off must leave the request byte-identical"
898        );
899    }
900
901    #[test]
902    fn cache_breakpoint_respects_client_anchor() {
903        // #939 safety: a client cache_control on a message means `system` is part
904        // of the already-cached prefix — never add a second, prefix-shifting anchor.
905        let _iso = crate::core::data_dir::isolated_data_dir();
906        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
907        let body = serde_json::json!({
908            "model": "claude-opus-4-8",
909            "system": cacheable_system(),
910            "messages": [{
911                "role": "user",
912                "content": [{
913                    "type": "text",
914                    "text": "hello",
915                    "cache_control": {"type": "ephemeral"}
916                }]
917            }]
918        });
919        let bytes = serde_json::to_vec(&body).unwrap();
920        crate::core::config::Config::update_global(|c| c.proxy.cache_breakpoint = Some(true))
921            .unwrap();
922        let (out, _o, _c) = compress_request_body(body, bytes.len());
923        let v: Value = serde_json::from_slice(&out).unwrap();
924        assert!(
925            v["system"].is_string(),
926            "with a client anchor present, system must be left untouched (no second breakpoint)"
927        );
928    }
929
930    #[test]
931    fn cache_aligner_measures_without_mutating_body() {
932        // #940: the volatile-field scan is telemetry-only — enabling it must leave
933        // the request byte-identical (measurement, not a rewrite), even on a
934        // volatile-field-rich system prompt.
935        let _iso = crate::core::data_dir::isolated_data_dir();
936        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_ALIGNER");
937        crate::test_env::remove_var("LEAN_CTX_PROXY_CACHE_BREAKPOINT");
938        let body = serde_json::json!({
939            "model": "claude-opus-4-8",
940            "system": "Today is 2026-06-22. Session 550e8400-e29b-41d4-a716-446655440000.",
941            "messages": [{"role": "user", "content": "Hello."}]
942        });
943        let bytes = serde_json::to_vec(&body).unwrap();
944        crate::core::config::Config::update_global(|c| c.proxy.cache_aligner = Some(true)).unwrap();
945        let (out, _o, _c) = compress_request_body(body, bytes.len());
946        assert_eq!(
947            out, bytes,
948            "cache-aligner telemetry must never mutate the request body"
949        );
950    }
951}