From 6f0b08b579a110de91e3bc259061dca6d2f9999a Mon Sep 17 00:00:00 2001 From: mnajafian-nv Date: Thu, 24 Sep 2026 23:36:36 -0700 Subject: [PATCH 1/2] feat: cache safe Anthropic on-demand compaction responses Signed-off-by: mnajafian-nv --- .../adaptive/src/response_cache/anthropic.rs | 106 +++++++- .../tests/integration/response_cache_tests.rs | 232 +++++++++++++++++- .../unit/response_cache/anthropic_tests.rs | 126 ++++++++-- .../tests/unit/response_cache/key_tests.rs | 13 +- .../tests/unit/response_cache/mark_tests.rs | 4 +- 5 files changed, 432 insertions(+), 49 deletions(-) diff --git a/crates/adaptive/src/response_cache/anthropic.rs b/crates/adaptive/src/response_cache/anthropic.rs index 7e36d6093..6e6a43db0 100644 --- a/crates/adaptive/src/response_cache/anthropic.rs +++ b/crates/adaptive/src/response_cache/anthropic.rs @@ -12,6 +12,7 @@ use serde_json::{Value as Json, json}; use crate::response_cache::mark::CacheReason; const THRESHOLD_BETA: &str = "compact-2026-01-12"; +const ON_DEMAND_BETA: &str = "compact-2026-09-04"; const ANTHROPIC_API_VERSION: &str = "2023-06-01"; const MIN_THRESHOLD_TOKENS: u64 = 50_000; const MAX_INSTRUCTIONS_CHARS: usize = 16_384; @@ -19,6 +20,7 @@ const MAX_INSTRUCTIONS_CHARS: usize = 16_384; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] pub(crate) enum AnthropicProtocol { ThresholdV1, + OnDemandV1, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] @@ -26,12 +28,16 @@ pub(crate) enum AnthropicOperation { PausedThreshold, ThresholdContinuation, ThresholdRecompact, + OnDemandSummarize, + OnDemandContinuation, + OnDemandRecompact, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum AnthropicResponseKind { Ordinary, ThresholdCompaction, + OnDemandCompaction, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -96,6 +102,7 @@ pub(crate) fn cache_context( .collect(); let protocol = match compact_tokens.as_slice() { [token] if *token == THRESHOLD_BETA && !on_demand_related => AnthropicProtocol::ThresholdV1, + [token] if *token == ON_DEMAND_BETA && !threshold_related => AnthropicProtocol::OnDemandV1, _ => return Err(CacheReason::AnthropicCompaction), }; @@ -106,7 +113,14 @@ pub(crate) fn cache_context( { return Err(CacheReason::AnthropicCompaction); } - if continuation && !has_supported_continuation(body) { + if on_demand_related + && !body + .get("compaction") + .is_some_and(is_supported_on_demand_request) + { + return Err(CacheReason::AnthropicCompaction); + } + if continuation && !has_supported_continuation(body, protocol) { return Err(CacheReason::AnthropicCompaction); } @@ -118,6 +132,13 @@ pub(crate) fn cache_context( (AnthropicProtocol::ThresholdV1, true, false, true) => { AnthropicOperation::ThresholdRecompact } + (AnthropicProtocol::OnDemandV1, false, true, false) => { + AnthropicOperation::OnDemandSummarize + } + (AnthropicProtocol::OnDemandV1, false, false, true) => { + AnthropicOperation::OnDemandContinuation + } + (AnthropicProtocol::OnDemandV1, false, true, true) => AnthropicOperation::OnDemandRecompact, _ => return Err(CacheReason::AnthropicCompaction), }; @@ -221,6 +242,16 @@ fn context_management_is_compaction_related(value: &Json) -> bool { }) } +fn is_supported_on_demand_request(value: &Json) -> bool { + value.as_object().is_some_and(|object| { + object + .keys() + .all(|key| matches!(key.as_str(), "type" | "instructions")) + && object.get("type").and_then(Json::as_str) == Some("summarize") + && object.get("instructions").is_none_or(valid_instructions) + }) +} + fn compaction_blocks(body: &serde_json::Map) -> Vec<&Json> { body.get("messages") .and_then(Json::as_array) @@ -236,12 +267,25 @@ fn contains_compaction_block(body: &serde_json::Map) -> bool { !compaction_blocks(body).is_empty() } -fn has_supported_continuation(body: &serde_json::Map) -> bool { +fn has_supported_continuation( + body: &serde_json::Map, + protocol: AnthropicProtocol, +) -> bool { let blocks = compaction_blocks(body); - !blocks.is_empty() - && blocks - .iter() - .all(|block| threshold_request_block_is_valid(block)) + match protocol { + AnthropicProtocol::ThresholdV1 => { + !blocks.is_empty() + && blocks + .iter() + .all(|block| threshold_request_block_is_valid(block)) + } + AnthropicProtocol::OnDemandV1 => { + let [block] = blocks.as_slice() else { + return false; + }; + on_demand_request_block_is_valid(block) + } + } } fn cache_control_is_valid(value: &Json) -> bool { @@ -263,6 +307,18 @@ fn threshold_request_block_is_valid(block: &Json) -> bool { .is_none_or(cache_control_is_valid) } +fn on_demand_request_block_is_valid(block: &Json) -> bool { + has_only_fields(block, &["type", "content", "signature", "cache_control"]) + && block.get("content").is_some_and(Json::is_string) + && block + .get("signature") + .and_then(Json::as_str) + .is_some_and(|value| !value.is_empty()) + && block + .get("cache_control") + .is_none_or(cache_control_is_valid) +} + pub(crate) fn classify_aggregate( response: &Json, context: &AnthropicCacheContext, @@ -335,6 +391,12 @@ pub(crate) fn classify_aggregate( { Some(AnthropicResponseKind::ThresholdCompaction) } + AnthropicProtocol::OnDemandV1 + if operation_allows_on_demand_compaction(context.operation) + && on_demand_response_block_is_valid(block) => + { + Some(AnthropicResponseKind::OnDemandCompaction) + } _ => None, } } @@ -345,6 +407,7 @@ fn operation_allows_ordinary(operation: AnthropicOperation) -> bool { AnthropicOperation::PausedThreshold | AnthropicOperation::ThresholdContinuation | AnthropicOperation::ThresholdRecompact + | AnthropicOperation::OnDemandContinuation ) } @@ -355,6 +418,13 @@ fn operation_allows_threshold_compaction(operation: AnthropicOperation) -> bool ) } +fn operation_allows_on_demand_compaction(operation: AnthropicOperation) -> bool { + matches!( + operation, + AnthropicOperation::OnDemandSummarize | AnthropicOperation::OnDemandRecompact + ) +} + fn valid_compaction_iterations(value: &Json) -> bool { value.as_array().is_some_and(|iterations| { !iterations.is_empty() @@ -379,6 +449,15 @@ fn threshold_response_block_is_valid(block: &Json) -> bool { .is_none_or(|value| value.is_null() || value.is_string()) } +fn on_demand_response_block_is_valid(block: &Json) -> bool { + has_only_fields(block, &["type", "content", "signature"]) + && block.get("content").is_some_and(Json::is_string) + && block + .get("signature") + .and_then(Json::as_str) + .is_some_and(|value| !value.is_empty()) +} + fn has_only_fields(value: &Json, allowed: &[&str]) -> bool { value .as_object() @@ -471,6 +550,12 @@ impl AnthropicStreamValidator { && compaction.len() == 1 && compaction[0].compaction_delta_count == 1 } + AnthropicResponseKind::OnDemandCompaction => { + self.protocol == AnthropicProtocol::OnDemandV1 + && self.blocks.len() == 1 + && compaction.len() == 1 + && compaction[0].compaction_delta_count == 0 + } } } @@ -542,8 +627,13 @@ impl AnthropicStreamValidator { return false; }; if kind == "compaction" { - let valid = has_only_fields(block, &["type", "content"]) - && block.get("content").is_some_and(Json::is_null); + let valid = match self.protocol { + AnthropicProtocol::ThresholdV1 => { + has_only_fields(block, &["type", "content"]) + && block.get("content").is_some_and(Json::is_null) + } + AnthropicProtocol::OnDemandV1 => on_demand_response_block_is_valid(block), + }; if !valid { return false; } diff --git a/crates/adaptive/tests/integration/response_cache_tests.rs b/crates/adaptive/tests/integration/response_cache_tests.rs index 264675db3..14da8004b 100644 --- a/crates/adaptive/tests/integration/response_cache_tests.rs +++ b/crates/adaptive/tests/integration/response_cache_tests.rs @@ -1395,6 +1395,18 @@ fn anthropic_compaction_stream_chunks() -> Vec { ] } +fn anthropic_on_demand_compaction_stream_chunks() -> Vec { + vec![ + json!({"type": "ping"}), + json!({"type": "message_start", "message": {"id": "msg_compact", "type": "message", "role": "assistant", "model": "claude-opus-5-5", "content": [], "container": null, "stop_reason": null, "stop_sequence": null, "stop_details": null, "diagnostics": null, "service_tier": "standard", "usage": {"input_tokens": 0, "output_tokens": 0}}}), + json!({"type": "content_block_start", "index": 0, "content_block": {"type": "compaction", "content": "summary", "signature": "signed"}}), + json!({"type": "ping"}), + json!({"type": "content_block_stop", "index": 0}), + json!({"type": "message_delta", "delta": {"stop_reason": "compaction", "stop_sequence": null, "stop_details": null, "container": null}, "usage": {"input_tokens": 0, "output_tokens": 0, "iterations": [{"type": "compaction", "input_tokens": 10, "output_tokens": 5}]}}), + json!({"type": "message_stop"}), + ] +} + fn anthropic_compaction_body() -> Json { json!({ "id": "msg_compact", @@ -1419,6 +1431,29 @@ fn anthropic_compaction_body() -> Json { }) } +fn anthropic_on_demand_compaction_body() -> Json { + json!({ + "id": "msg_compact", + "type": "message", + "role": "assistant", + "model": "claude-opus-5-5", + "container": null, + "stop_details": null, + "diagnostics": null, + "service_tier": "standard", + "content": [{ + "type": "compaction", "content": "summary", "signature": "signed" + }], + "stop_reason": "compaction", + "stop_sequence": null, + "usage": { + "input_tokens": 0, + "output_tokens": 0, + "iterations": [{"type": "compaction", "input_tokens": 10, "output_tokens": 5}] + } + }) +} + /// Like [`stream_call`], but with a caller-chosen provider name (the value the /// gateway derives from the route, e.g. `"anthropic.messages"`). async fn stream_call_named( @@ -2134,13 +2169,16 @@ async fn threshold_compaction_with_a_sparse_index_is_never_stored() { } #[tokio::test] -async fn on_demand_compaction_remains_live() { +async fn on_demand_compaction_replays_signed_block_without_delta() { let _guard = TEST_MUTEX.lock().await; reset_global(); activate_cache(compaction_cache_config()).await; let calls = Arc::new(AtomicUsize::new(0)); - let provider = counting_stream_provider(Arc::clone(&calls), anthropic_stream_chunks()); + let provider = counting_stream_provider( + Arc::clone(&calls), + anthropic_on_demand_compaction_stream_chunks(), + ); let request = || LlmRequest { headers: serde_json::Map::from_iter([ ("anthropic-version".to_string(), json!("2023-06-01")), @@ -2154,13 +2192,62 @@ async fn on_demand_compaction_remains_live() { }; stream_call_named("anthropic.messages", &provider, request()).await; - stream_call_named("anthropic.messages", &provider, request()).await; + let replayed = stream_call_named("anthropic.messages", &provider, request()).await; assert_eq!( calls.load(Ordering::SeqCst), - 2, - "on-demand compaction must remain live until its protocol is supported" + 1, + "validated on-demand response must hit" ); + assert!(replayed.iter().any(|chunk| { + chunk + .pointer("/content_block/signature") + .and_then(Json::as_str) + == Some("signed") + })); + assert!(!replayed.iter().any(|chunk| { + chunk.pointer("/delta/type").and_then(Json::as_str) == Some("compaction_delta") + })); +} + +#[tokio::test] +async fn on_demand_recompaction_replays_the_new_signed_block() { + let _guard = TEST_MUTEX.lock().await; + reset_global(); + activate_cache(compaction_cache_config()).await; + + let calls = Arc::new(AtomicUsize::new(0)); + let provider = counting_stream_provider( + Arc::clone(&calls), + anthropic_on_demand_compaction_stream_chunks(), + ); + let request = || LlmRequest { + headers: serde_json::Map::from_iter([ + ("anthropic-version".to_string(), json!("2023-06-01")), + ("anthropic-beta".to_string(), json!("compact-2026-09-04")), + ]), + content: json!({ + "model": "claude-opus-5-5", + "messages": [ + {"role": "assistant", "content": [{ + "type": "compaction", "content": "old summary", "signature": "old-signed" + }]}, + {"role": "user", "content": "continue"} + ], + "compaction": {"type": "summarize"} + }), + }; + + stream_call_named("anthropic.messages", &provider, request()).await; + let replayed = stream_call_named("anthropic.messages", &provider, request()).await; + + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert!(replayed.iter().any(|chunk| { + chunk + .pointer("/content_block/signature") + .and_then(Json::as_str) + == Some("signed") + })); } #[tokio::test] @@ -2224,6 +2311,38 @@ async fn paused_threshold_ordinary_stream_with_unknown_delta_field_is_never_stor assert_eq!(calls.load(Ordering::SeqCst), 2); } +#[tokio::test] +async fn on_demand_continuation_caches_ordinary_response() { + let _guard = TEST_MUTEX.lock().await; + reset_global(); + activate_cache(compaction_cache_config()).await; + + let calls = Arc::new(AtomicUsize::new(0)); + let provider = counting_stream_provider(Arc::clone(&calls), anthropic_stream_chunks()); + let request = || LlmRequest { + headers: serde_json::Map::from_iter([ + ("anthropic-version".to_string(), json!("2023-06-01")), + ("anthropic-beta".to_string(), json!("compact-2026-09-04")), + ]), + content: json!({ + "model": "claude-opus-5-5", + "messages": [ + {"role": "assistant", "content": [{ + "type": "compaction", "content": "summary", "signature": "signed", + "cache_control": {"type": "ephemeral"} + }]}, + {"role": "user", "content": "continue"} + ] + }), + }; + + stream_call_named("anthropic.messages", &provider, request()).await; + let replayed = stream_call_named("anthropic.messages", &provider, request()).await; + + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(replayed_text(&replayed), "Hello, world."); +} + #[tokio::test] async fn threshold_continuation_preserves_null_encrypted_content_and_caches_ordinary_response() { let _guard = TEST_MUTEX.lock().await; @@ -2381,6 +2500,109 @@ async fn threshold_compaction_cache_entry_is_shared_across_buffered_and_streamin })); } +#[tokio::test] +async fn on_demand_compaction_cache_entry_is_shared_across_buffered_and_streaming_calls() { + let _guard = TEST_MUTEX.lock().await; + let request = || LlmRequest { + headers: serde_json::Map::from_iter([ + ("anthropic-version".to_string(), json!("2023-06-01")), + ("anthropic-beta".to_string(), json!("compact-2026-09-04")), + ]), + content: json!({ + "model": "claude-opus-5-5", + "messages": [{"role": "user", "content": "hello"}], + "compaction": {"type": "summarize"} + }), + }; + + reset_global(); + activate_cache(compaction_cache_config()).await; + let stream_calls = Arc::new(AtomicUsize::new(0)); + let stream_provider = counting_stream_provider( + Arc::clone(&stream_calls), + anthropic_on_demand_compaction_stream_chunks(), + ); + stream_call_named("anthropic.messages", &stream_provider, request()).await; + let buffered_calls = Arc::new(AtomicUsize::new(0)); + let buffered_provider = counting_provider( + Arc::clone(&buffered_calls), + json!({"unexpected": "provider must not run"}), + ); + let buffered = buffered_call_named("anthropic.messages", &buffered_provider, request()).await; + assert_eq!(stream_calls.load(Ordering::SeqCst), 1); + assert_eq!(buffered_calls.load(Ordering::SeqCst), 0); + assert_eq!(buffered, anthropic_on_demand_compaction_body()); + + reset_global(); + activate_cache(compaction_cache_config()).await; + let buffered_calls = Arc::new(AtomicUsize::new(0)); + let buffered_provider = counting_provider( + Arc::clone(&buffered_calls), + anthropic_on_demand_compaction_body(), + ); + buffered_call_named("anthropic.messages", &buffered_provider, request()).await; + let stream_calls = Arc::new(AtomicUsize::new(0)); + let stream_provider = counting_stream_provider(Arc::clone(&stream_calls), vec![]); + let replayed = stream_call_named("anthropic.messages", &stream_provider, request()).await; + assert_eq!(buffered_calls.load(Ordering::SeqCst), 1); + assert_eq!(stream_calls.load(Ordering::SeqCst), 0); + assert!(replayed.iter().any(|chunk| { + chunk + .pointer("/content_block/signature") + .and_then(Json::as_str) + == Some("signed") + })); + assert!(!replayed.iter().any(|chunk| { + chunk.pointer("/delta/type").and_then(Json::as_str) == Some("compaction_delta") + })); +} + +#[tokio::test] +async fn malformed_buffered_on_demand_response_is_never_stored() { + let _guard = TEST_MUTEX.lock().await; + reset_global(); + activate_cache(compaction_cache_config()).await; + + let calls = Arc::new(AtomicUsize::new(0)); + let provider = counting_provider( + Arc::clone(&calls), + json!({ + "id": "msg_compact", + "type": "message", + "role": "assistant", + "model": "claude-opus-5-5", + "content": [{"type": "compaction", "content": "summary"}], + "stop_reason": "compaction", + "stop_sequence": null, + "usage": { + "input_tokens": 0, + "output_tokens": 0, + "iterations": [{"type": "compaction", "input_tokens": 10, "output_tokens": 5}] + } + }), + ); + let request = || LlmRequest { + headers: serde_json::Map::from_iter([ + ("anthropic-version".to_string(), json!("2023-06-01")), + ("anthropic-beta".to_string(), json!("compact-2026-09-04")), + ]), + content: json!({ + "model": "claude-opus-5-5", + "messages": [{"role": "user", "content": "hello"}], + "compaction": {"type": "summarize"} + }), + }; + + buffered_call_named("anthropic.messages", &provider, request()).await; + buffered_call_named("anthropic.messages", &provider, request()).await; + + assert_eq!( + calls.load(Ordering::SeqCst), + 2, + "an unsigned on-demand block must remain live" + ); +} + #[tokio::test] async fn no_codec_buffered_and_streaming_share_one_store_entry() { // Buffered and streaming derive the same key from the auto-detected decode, diff --git a/crates/adaptive/tests/unit/response_cache/anthropic_tests.rs b/crates/adaptive/tests/unit/response_cache/anthropic_tests.rs index 4d1a4690c..e66d95c52 100644 --- a/crates/adaptive/tests/unit/response_cache/anthropic_tests.rs +++ b/crates/adaptive/tests/unit/response_cache/anthropic_tests.rs @@ -32,6 +32,10 @@ fn threshold_block() -> Json { json!({"type": "compaction", "content": "summary", "encrypted_content": null}) } +fn on_demand_block() -> Json { + json!({"type": "compaction", "content": "summary", "signature": "signed"}) +} + fn ordinary_response() -> Json { json!({ "id": "msg_ordinary", @@ -62,15 +66,36 @@ fn threshold_response() -> Json { }) } +fn on_demand_response() -> Json { + json!({ + "id": "msg_on_demand", + "type": "message", + "role": "assistant", + "model": "claude-opus-5-5", + "container": null, + "stop_details": null, + "diagnostics": null, + "service_tier": "standard", + "content": [on_demand_block()], + "stop_reason": "compaction", + "stop_sequence": null, + "usage": { + "input_tokens": 0, + "output_tokens": 0, + "iterations": [{"type": "compaction", "input_tokens": 10, "output_tokens": 5}] + } + }) +} + #[test] fn unknown_null_response_metadata_is_not_cacheable() { let context = AnthropicCacheContext { version: ANTHROPIC_API_VERSION.to_string(), - beta_tokens: vec![THRESHOLD_BETA.to_string()], - protocol: AnthropicProtocol::ThresholdV1, - operation: AnthropicOperation::PausedThreshold, + beta_tokens: vec![ON_DEMAND_BETA.to_string()], + protocol: AnthropicProtocol::OnDemandV1, + operation: AnthropicOperation::OnDemandSummarize, }; - let mut response = threshold_response(); + let mut response = on_demand_response(); response["future_response_field"] = Json::Null; assert_eq!(classify_aggregate(&response, &context), None); @@ -114,6 +139,41 @@ fn request_shapes_map_to_distinct_compaction_operations() { ), AnthropicOperation::ThresholdRecompact, ), + ( + request( + json!({ + "messages": [{"role": "user", "content": "hello"}], + "compaction": {"type": "summarize"} + }), + ON_DEMAND_BETA, + ), + AnthropicOperation::OnDemandSummarize, + ), + ( + request( + json!({ + "messages": [ + {"role": "assistant", "content": [on_demand_block()]}, + {"role": "user", "content": "continue"} + ] + }), + ON_DEMAND_BETA, + ), + AnthropicOperation::OnDemandContinuation, + ), + ( + request( + json!({ + "messages": [ + {"role": "assistant", "content": [on_demand_block()]}, + {"role": "user", "content": "continue"} + ], + "compaction": {"type": "summarize"} + }), + ON_DEMAND_BETA, + ), + AnthropicOperation::OnDemandRecompact, + ), ]; for (request, expected) in cases { @@ -128,61 +188,77 @@ fn request_shapes_map_to_distinct_compaction_operations() { fn response_kind_must_match_the_request_operation() { let ordinary = ordinary_response(); let threshold = threshold_response(); + let on_demand = on_demand_response(); let cases = [ ( AnthropicOperation::PausedThreshold, Some(AnthropicResponseKind::Ordinary), Some(AnthropicResponseKind::ThresholdCompaction), + None, ), ( AnthropicOperation::ThresholdContinuation, Some(AnthropicResponseKind::Ordinary), None, + None, ), ( AnthropicOperation::ThresholdRecompact, Some(AnthropicResponseKind::Ordinary), Some(AnthropicResponseKind::ThresholdCompaction), + None, + ), + ( + AnthropicOperation::OnDemandSummarize, + None, + None, + Some(AnthropicResponseKind::OnDemandCompaction), + ), + ( + AnthropicOperation::OnDemandContinuation, + Some(AnthropicResponseKind::Ordinary), + None, + None, + ), + ( + AnthropicOperation::OnDemandRecompact, + None, + None, + Some(AnthropicResponseKind::OnDemandCompaction), ), ]; - for (operation, ordinary_kind, threshold_kind) in cases { + for (operation, ordinary_kind, threshold_kind, on_demand_kind) in cases { + let protocol = match operation { + AnthropicOperation::OnDemandSummarize + | AnthropicOperation::OnDemandContinuation + | AnthropicOperation::OnDemandRecompact => AnthropicProtocol::OnDemandV1, + _ => AnthropicProtocol::ThresholdV1, + }; let context = AnthropicCacheContext { version: ANTHROPIC_API_VERSION.to_string(), - beta_tokens: vec![THRESHOLD_BETA.to_string()], - protocol: AnthropicProtocol::ThresholdV1, + beta_tokens: vec![match protocol { + AnthropicProtocol::ThresholdV1 => THRESHOLD_BETA.to_string(), + AnthropicProtocol::OnDemandV1 => ON_DEMAND_BETA.to_string(), + }], + protocol, operation, }; assert_eq!(classify_aggregate(&ordinary, &context), ordinary_kind); assert_eq!(classify_aggregate(&threshold, &context), threshold_kind); + assert_eq!(classify_aggregate(&on_demand, &context), on_demand_kind); } } -#[test] -fn on_demand_compaction_remains_live() { - let request = request( - json!({ - "messages": [{"role": "user", "content": "hello"}], - "compaction": {"type": "summarize"} - }), - "compact-2026-09-04", - ); - - assert_eq!( - cache_context(&request), - Err(CacheReason::AnthropicCompaction) - ); -} - #[test] fn unsupported_anthropic_api_versions_bypass_compaction_caching() { let mut request = request( json!({ "messages": [{"role": "user", "content": "hello"}], - "context_management": threshold_edit() + "compaction": {"type": "summarize"} }), - THRESHOLD_BETA, + ON_DEMAND_BETA, ); request .headers diff --git a/crates/adaptive/tests/unit/response_cache/key_tests.rs b/crates/adaptive/tests/unit/response_cache/key_tests.rs index 3fd1037fd..0b2367168 100644 --- a/crates/adaptive/tests/unit/response_cache/key_tests.rs +++ b/crates/adaptive/tests/unit/response_cache/key_tests.rs @@ -653,17 +653,12 @@ fn ambiguous_anthropic_protocol_headers_bypass() { let mut request = LlmRequest { headers: Map::from_iter([ ("anthropic-version".to_string(), json!("2023-06-01")), - ("anthropic-beta".to_string(), json!("compact-2026-01-12")), + ("anthropic-beta".to_string(), json!("compact-2026-09-04")), ]), content: json!({ "model": "claude-opus-5-5", "messages": [{"role": "user", "content": "hello"}], - "context_management": { - "edits": [{ - "type": "compact_20260112", - "pause_after_compaction": true - }] - } + "compaction": {"type": "summarize"} }), }; request @@ -677,10 +672,10 @@ fn ambiguous_anthropic_protocol_headers_bypass() { } #[test] -fn on_demand_compaction_keeps_protocol_bypass_reason_by_default() { +fn unsupported_compaction_protocol_keeps_protocol_bypass_reason_by_default() { let request = LlmRequest { headers: Map::from_iter([ - ("anthropic-version".to_string(), json!("2023-06-01")), + ("anthropic-version".to_string(), json!("2024-01-01")), ("anthropic-beta".to_string(), json!("compact-2026-09-04")), ]), content: json!({ diff --git a/crates/adaptive/tests/unit/response_cache/mark_tests.rs b/crates/adaptive/tests/unit/response_cache/mark_tests.rs index 14633be97..5cdd388e5 100644 --- a/crates/adaptive/tests/unit/response_cache/mark_tests.rs +++ b/crates/adaptive/tests/unit/response_cache/mark_tests.rs @@ -128,7 +128,7 @@ fn anthropic_shaped_bodies_price_through_the_catalog() { "content": [{ "type": "compaction", "content": "summary", - "encrypted_content": null + "signature": "signed" }], "stop_reason": "compaction", "usage": { @@ -190,7 +190,7 @@ fn compaction_savings_include_iteration_tokens() { "content": [{ "type": "compaction", "content": "summary", - "encrypted_content": null + "signature": "signed" }], "stop_reason": "compaction", "stop_sequence": null, From 8733225e3848801ab6e4a93def8b4ca72ad086d9 Mon Sep 17 00:00:00 2001 From: mnajafian-nv Date: Thu, 24 Sep 2026 23:39:33 -0700 Subject: [PATCH 2/2] docs: document Anthropic on-demand compaction cache support Signed-off-by: mnajafian-nv --- docs/configure-plugins/adaptive/response-cache.mdx | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/docs/configure-plugins/adaptive/response-cache.mdx b/docs/configure-plugins/adaptive/response-cache.mdx index b44d75d1c..b6c08b80a 100644 --- a/docs/configure-plugins/adaptive/response-cache.mdx +++ b/docs/configure-plugins/adaptive/response-cache.mdx @@ -314,7 +314,11 @@ Only complete, replayable LLM answers are stored: `type = "compaction"` block, string `content`, and omitted, null, or string `encrypted_content` can cache an ordinary response and can trigger another cached compaction when it also carries the paused threshold edit. On-demand - compaction requests remain live. + compaction requires the `compact-2026-09-04` beta and + `compaction: {"type": "summarize"}`; Relay replays its single signed block + without synthesizing a delta. A valid on-demand continuation has one signed + `type = "compaction"` block and can cache an ordinary response or trigger + another cached compaction when it also carries a new `summarize` request. Version and ordered beta tokens partition these entries from ordinary Messages requests. Extra response-bearing fields, unknown events, sparse or out-of-order blocks, and missing, ambiguous, unsupported, or malformed