From 2edaa36835a65fec40027d44322731f301c9987c Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 29 Oct 2026 23:27:11 +0000 Subject: [PATCH] Pin the compaction summary at the request's cache lifetime On a bot set to the hour-long Anthropic cache, the pinned compaction summary and carry-forward note kept a breakpoint with no lifetime, which the API reads as five minutes. Instructions (1h), then the summary (5m), then the request's own breakpoint (1h) breaks the rule that a longer lifetime may not follow a shorter one, so every call after a bot's first summary was refused with a 400. Every breakpoint in a request now takes one lifetime, as other harnesses compute one cache marker per request and use it everywhere. The turn picks it once: the bot's setting, unless the turn's model, its summarizer (which copies the turn's calls) or its fallback (which answers them) must keep five minutes, in which case they all do. The pinned blocks are rendered from the stored summary on every request, so a bot already stuck this way recovers on its next turn with no repair. The Anthropic test fixture now refuses misordered lifetimes as the API does; the new compaction tests fail without the fix. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013jkdGEPbASx5sbXvPAfKHw --- docs/RUST_PROTOTYPE.md | 2 +- src/codec.rs | 11 ++++ src/provider.rs | 17 ++--- src/server/turn.rs | 131 +++++++++++++++++++++++++++++++++++---- src/store/context.rs | 30 ++++++--- src/store/db.rs | 3 + tests/store_contract.rs | 2 +- tests/test_compaction.py | 52 ++++++++++++++++ tests/test_runtime.py | 15 +++++ 9 files changed, 236 insertions(+), 27 deletions(-) diff --git a/docs/RUST_PROTOTYPE.md b/docs/RUST_PROTOTYPE.md index 626a7b4b9..7b6bf15fa 100644 --- a/docs/RUST_PROTOTYPE.md +++ b/docs/RUST_PROTOTYPE.md @@ -408,7 +408,7 @@ keep-warm when it sets any of them. | `approval_hold_ms` (`--approval-hold`) | Milliseconds a gated call waits live for its verdict before its turn parks (see [tool approval](#tool-approval)). 0 parks at once; at most 3,600,000. | 2,000 | | `max_output_tokens` (`--max-output-tokens`) | Generated tokens per model call, including reasoning, summaries included. Anthropic calls use the model's full output limit (read inside Bedrock ids) unless this is set; set, it is sent as `max_tokens`, at least 2,048 so a legacy thinking budget of 1,024 or more fits beside the answer. Bedrock deducts input plus this bound from quota when a call starts, so a bound near real output buys throughput there. | none | | `keep_warm` (`--keep-warm`) | Seconds an Anthropic prompt cache may sit unread while a turn runs a tool before it is refreshed (see [keeping the cache warm](#keeping-the-anthropic-cache-warm)). Below 300; 0 never refreshes. | 240 | -| `cache_ttl` (`--cache-ttl`) | Anthropic prompt-cache lifetime, `5m` or `1h`, on both cache markers. `1h` bills each write at twice the input rate instead of 1.25 times and sends no refreshes. Responses providers are unaffected. Bedrock takes `1h` for Claude 4.5 and later and may refuse it for older Claude (its prompt-caching guide, read 2026-10-28), so calls to older Claude there keep five minutes whatever this says, also after `configure` switches a bot to one. | `5m` | +| `cache_ttl` (`--cache-ttl`) | Anthropic prompt-cache lifetime, `5m` or `1h`, on every cache marker of a request: the instructions', the request's own, and the pinned compaction summary's and note's, since the API refuses a longer lifetime after a shorter one. `1h` bills each write at twice the input rate instead of 1.25 times and sends no refreshes. Responses providers are unaffected. Bedrock takes `1h` for Claude 4.5 and later and may refuse it for older Claude (its prompt-caching guide, read 2026-10-28), so calls to older Claude there keep five minutes whatever this says, also after `configure` switches a bot to one. A bot whose summarizer or fallback is such a model keeps five minutes on every call, since those calls carry the same pinned blocks. | `5m` | | `sandbox` (`--sandbox`) | How the bot's commands, writes and edits are confined, by Codex's names: `read-only`, `workspace-write` or `danger-full-access` (see [sandbox](#sandbox)). | none | | `writable` (`--writable`, repeatable) | Absolute folders a `workspace-write` bot may write besides its own; each must exist and is stored by its real path. At most 64, 32 KiB as JSON. | none | | `network` (`--network`) | A `workspace-write` bot's commands may use the network. | off | diff --git a/src/codec.rs b/src/codec.rs index 33b7ad623..495ef1ece 100644 --- a/src/codec.rs +++ b/src/codec.rs @@ -5,6 +5,17 @@ use crate::{Result, fail}; use base64::Engine; use serde_json::{Value, json}; +/// An Anthropic cache breakpoint, an hour's or five minutes'. Every +/// breakpoint in one request takes the same lifetime: the API refuses a +/// longer one after a shorter one, and the request's own last breakpoint +/// comes after every other. +pub fn cache_control(hour: bool) -> Value { + match hour { + true => json!({"type":"ephemeral","ttl":"1h"}), + false => json!({"type":"ephemeral"}), + } +} + #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)] pub enum Family { /// OpenAI Responses API and compatible gateways. diff --git a/src/provider.rs b/src/provider.rs index 443594c85..b17a0f163 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -631,7 +631,13 @@ impl Provider { /// five minutes whatever the bot's setting, including after a switch to /// such a model. fn cache_hour_for(&self, model: &str) -> bool { - self.cache_hour && !(self.bedrock && five_minute_cache(model)) + self.cache_hour && self.takes_cache_hour(model) + } + + /// Whether a call to `model` here may ask for the hour-long cache at + /// all, whatever a bot's setting. + pub fn takes_cache_hour(&self, model: &str) -> bool { + self.family == Family::Anthropic && !(self.bedrock && five_minute_cache(model)) } /// This provider with one bot's output bound, cache lifetime and @@ -732,12 +738,9 @@ impl Provider { .then(|| self.search(request.model).entry(self.family)) .flatten(); let max_tokens = self.anthropic_max_tokens(request.model); - // Both markers take the same lifetime: a longer one may not follow - // a shorter one. - let cache = match self.cache_hour_for(request.model) { - true => json!({"type":"ephemeral","ttl":"1h"}), - false => json!({"type":"ephemeral"}), - }; + // Every marker takes the same lifetime, the pinned blocks' among + // them: a longer one may not follow a shorter one. + let cache = crate::codec::cache_control(self.cache_hour_for(request.model)); // The ChatGPT plan takes function tools only grouped in a namespace // (OpenAI's preview limitations, read 2026-10-20); a call comes back // naming it, and its item is replayed as it came. Search is its own diff --git a/src/server/turn.rs b/src/server/turn.rs index abecd3eb3..6f9f58bb2 100644 --- a/src/server/turn.rs +++ b/src/server/turn.rs @@ -637,7 +637,27 @@ impl Turn { } else { Vec::new() }; - window.prefix(&listed, budget) + window.prefix(&listed, budget, self.settings().cache_hour()) + } + + /// The bot's settings, with one cache lifetime for every request that + /// carries its pinned summary and note: the turn's model's, the + /// summarizer's, which copies the turn's calls, and the fallback's, + /// which answers them. + fn with_one_lifetime(&self, record: &agent_runtime::store::Bot, model: &str) -> Settings { + let summarizer = summarizer(record); + one_lifetime( + &self.providers, + record.family().ok(), + &record.settings, + [ + Some(model), + Some(summarizer.as_str()), + record.fallback.as_deref(), + ] + .into_iter() + .flatten(), + ) } /// The bot's settings. Every use is inside `rounds`, which reads them @@ -1956,8 +1976,9 @@ impl Turn { ) -> Result { let (bot, turn) = (self.bot.clone(), self.turn); let mut record = self.store.op("inspect", move |db| db.inspect(&bot)).await?; - self.settings.get_or_init(|| record.settings.clone()); let context = self.store.op("context", move |db| db.context(turn)).await?; + self.settings + .get_or_init(|| self.with_one_lifetime(&record, &context.model)); // A turn may run at its own effort; everything after reads the record. record.effort = context.effort.clone(); record.turn_budget = context.budget; @@ -5162,6 +5183,28 @@ fn failure(error: Error) -> Outcome { } } +/// `settings` with five minutes for an hour where any of the `models` a +/// request may go to must keep five minutes, as a longer marker may not +/// follow a shorter one. A model on a provider unknown here or of another +/// family is sent nothing. +fn one_lifetime<'a>( + providers: &HashMap, + family: Option, + settings: &Settings, + models: impl IntoIterator, +) -> Settings { + let mut settings = settings.clone(); + let takes_hour = (models.into_iter()) + .filter_map(|reference| split_model(reference).ok()) + .filter_map(|(name, model)| Some((providers.get(name)?, model))) + .filter(|(provider, _)| Some(provider.family()) == family) + .all(|(provider, model)| provider.takes_cache_hour(model)); + if settings.cache_hour() && !takes_hour { + settings.cache_ttl = Some(agent_runtime::store::CacheTtl::Minutes); + } + settings +} + /// The summarizer: the bot's own model unless the client named one of the /// same family, checked at creation. fn summarizer(record: &agent_runtime::store::Bot) -> String { @@ -5216,20 +5259,86 @@ mod tests { assert!(super::backoff(1, 7).as_millis() < 400); } #[test] - fn pinned_blocks_cache_only_with_their_provider_format() { - use agent_runtime::codec::Family; - let anthropic: serde_json::Value = - serde_json::from_slice(&super::pinned_item(Family::Anthropic, "stable").unwrap()) - .unwrap(); + fn one_model_that_keeps_five_minutes_keeps_every_request_at_five() { + use super::{HashMap, Provider, Settings}; + use agent_runtime::{codec::Family, provider::Transport, store::CacheTtl}; + let provider = |url| { + Provider::new(Transport::new(1, 1).unwrap(), Family::Anthropic, url, None).unwrap() + }; + let providers = HashMap::from([ + ( + "anthropic".to_owned(), + provider("https://api.anthropic.com/v1"), + ), + ( + "bedrock".to_owned(), + provider("https://bedrock-runtime.us-east-1.amazonaws.com/anthropic/v1"), + ), + ]); + let hour = Settings { + cache_ttl: Some(CacheTtl::Hour), + ..Settings::default() + }; + let ttl = |models: &[&str]| { + super::one_lifetime( + &providers, + Some(Family::Anthropic), + &hour, + models.iter().copied(), + ) + .cache_ttl + }; + assert_eq!( + ttl(&[ + "anthropic/claude-opus-5-5", + "bedrock/us.anthropic.claude-opus-5-5-v1:0" + ]), + Some(CacheTtl::Hour) + ); + // A summarizer or fallback on older Claude there brings the turn's + // own calls down too; a provider not here is sent nothing. assert_eq!( - anthropic["content"][0]["cache_control"]["type"], - "ephemeral" + ttl(&[ + "anthropic/claude-opus-5-5", + "bedrock/anthropic.claude-sonnet-4-20250514-v1:0" + ]), + Some(CacheTtl::Minutes) ); assert_eq!( - super::pinned_item(Family::Responses, "stable").unwrap(), - Family::Responses.user_item("stable").unwrap() + ttl(&["anthropic/claude-opus-5-5", "gone/claude-3-7-sonnet"]), + Some(CacheTtl::Hour) + ); + // Five minutes stays five minutes. + assert_eq!( + super::one_lifetime( + &providers, + Some(Family::Anthropic), + &Settings::default(), + ["anthropic/claude-opus-5-5"] + ) + .cache_ttl, + None ); } + + #[test] + fn pinned_blocks_cache_only_with_their_provider_format() { + use agent_runtime::codec::Family; + for hour in [false, true] { + let anthropic: serde_json::Value = serde_json::from_slice( + &super::pinned_item(Family::Anthropic, "stable", hour).unwrap(), + ) + .unwrap(); + assert_eq!( + anthropic["content"][0]["cache_control"], + agent_runtime::codec::cache_control(hour) + ); + assert_eq!( + super::pinned_item(Family::Responses, "stable", hour).unwrap(), + Family::Responses.user_item("stable").unwrap() + ); + } + } use super::*; use serde_json::Value; diff --git a/src/store/context.rs b/src/store/context.rs index f55b7c283..54be27f17 100644 --- a/src/store/context.rs +++ b/src/store/context.rs @@ -1,6 +1,9 @@ //! Encoded context accounting shared by request construction and compaction. use super::{CompactionView, Window}; -use crate::{Result, codec::Family}; +use crate::{ + Result, + codec::{Family, cache_control}, +}; use serde::Serialize; use serde_json::json; @@ -48,7 +51,14 @@ impl Window { self.pictured.truncate(len); } - pub fn prefix(&self, listed: &[(i64, String)], prefix_budget: usize) -> Result { + /// What goes ahead of the window, its pinned blocks cached for an hour + /// when `hour`, as the request that carries them is. + pub fn prefix( + &self, + listed: &[(i64, String)], + prefix_budget: usize, + hour: bool, + ) -> Result { context_prefix( self.family, self.compaction.as_ref(), @@ -59,6 +69,7 @@ impl Window { self.history, listed, prefix_budget, + hour, ) } } @@ -70,7 +81,8 @@ impl CompactionView { /// Count each prompt once instead of repeatedly encoding the entire prefix. pub(crate) fn bound_prompt_bytes(&mut self, family: Family, budget: usize) -> Result<()> { let mut prompts = std::mem::take(&mut self.prompts); - let base = context_prefix(family, Some(self), None, 0, 0, 0, false, &[], 0)? + // With the longer marker, so the bound holds at either lifetime. + let base = context_prefix(family, Some(self), None, 0, 0, 0, false, &[], 0, true)? .bytes .len(); if base > budget { @@ -134,6 +146,7 @@ pub(crate) fn context_prefix( history: bool, listed: &[(i64, String)], prefix_budget: usize, + hour: bool, ) -> Result { let mut encoded = Vec::new(); let mut count = 0; @@ -160,12 +173,13 @@ pub(crate) fn context_prefix( text.push_str(&format!("\n{ordinal}: {prompt}")); } } - push(pinned_item(family, &text)?); + push(pinned_item(family, &text, hour)?); } if let Some((version, text)) = note.filter(|(_, text)| !text.is_empty()) { push(pinned_item( family, &format!("[carry-forward note, version {version}]\n{text}"), + hour, )?); } let pinned_bytes = encoded.len(); @@ -568,11 +582,12 @@ pub fn declined( Ok(None) } -/// Stable pinned blocks retain their own Anthropic cache breakpoints. -pub fn pinned_item(family: Family, text: &str) -> Result> { +/// Stable pinned blocks retain their own Anthropic cache breakpoints, at +/// the lifetime of the request they go in (`hour`). +pub fn pinned_item(family: Family, text: &str, hour: bool) -> Result> { match family { Family::Anthropic => Ok(serde_json::to_vec(&json!({ - "role":"user","content":[{"type":"text","text":text,"cache_control":{"type":"ephemeral"}}] + "role":"user","content":[{"type":"text","text":text,"cache_control":cache_control(hour)}] }))?), _ => family.user_item(text), } @@ -710,6 +725,7 @@ mod tests { history, &listed, 1 << 20, + false, ) .unwrap(); String::from_utf8(prefix.bytes.to_vec()).unwrap() diff --git a/src/store/db.rs b/src/store/db.rs index 0318caca2..c71be681f 100644 --- a/src/store/db.rs +++ b/src/store/db.rs @@ -4027,6 +4027,7 @@ impl Database { bot.tools.iter().any(|tool| tool == "history"), &listed, input_limit.bytes * 2 / 3, + bot.settings.cache_hour(), )?; Ok(super::ContextUsage { bytes: bytes as usize, @@ -4327,6 +4328,8 @@ impl Database { history, &[], 0, + // The longer marker, so the minimum is never under what is sent. + true, )?; Ok(( family, diff --git a/tests/store_contract.rs b/tests/store_contract.rs index 3e6b62d3a..9712419cd 100644 --- a/tests/store_contract.rs +++ b/tests/store_contract.rs @@ -10608,7 +10608,7 @@ fn an_elided_event_counts_what_the_move_takes_off_the_view() { /// The request prefix a window's view renders, as JSON items. fn prefix_items(window: &agent_runtime::store::Window) -> Vec { - let prefix = window.prefix(&[], usize::MAX).unwrap().bytes; + let prefix = window.prefix(&[], usize::MAX, false).unwrap().bytes; let prefix = &prefix[..prefix.len().saturating_sub(1)]; serde_json::from_slice(&[b"[", prefix, b"]"].concat()).unwrap() } diff --git a/tests/test_compaction.py b/tests/test_compaction.py index 92177d277..4245e7cfa 100644 --- a/tests/test_compaction.py +++ b/tests/test_compaction.py @@ -52,6 +52,58 @@ def test_a_summary_copies_the_call_before_it_with_its_tools_and_tool_choice(self self.assertIsNotNone(client.request('inspect', bot='Bob')['result']['compaction']) self.assertFalse(any(m.get('event') == 'compaction_failed' for m in client.saved)) + def test_an_hour_long_cache_pins_the_summary_for_an_hour_too(self): + # A pinned block at five minutes ahead of the request's own hour + # breakpoint is refused, so every turn after the first summary + # failed. Each later call marks the summary as the request does. + self.model.cache_control = {'type': 'ephemeral', 'ttl': '1h'} + client = Client(self.binary, self.path / 'state.sqlite', self.url, + tools='echo,shell', provider='anthropic', family='anthropic', + model='synthetic-claude', key_env='ANTHROPIC_TEST_KEY', + env={**clean_env(), 'ANTHROPIC_TEST_KEY': 'synthetic-anthropic-key'}, + settings={'context_bytes': 8192, 'compact_at': 50, 'cache_ttl': '1h'}) + self.addCleanup(client.close) + self.assertIn('result', client.request('create', bot='Bob', workspace=str(self.path), + compaction_instructions='Summarize.', effort='low')) + for n in range(7): + turn = client.request('submit', bot='Bob', request_id=str(n), prompt=f'{n}:' + 'x' * 500)['result']['turn'] + self.assertEqual(client.finished(turn)['data']['status'], 'completed') + self.assertIsNotNone(client.request('inspect', bot='Bob')['result']['compaction']) + requests = [] + while not self.model.requests.empty(): + requests.append(self.model.requests.get()) + pinned = [b for r in requests for m in r['messages'] for b in m['content'] + if b.get('text', '').startswith('[compaction summary')] + self.assertTrue(pinned) + self.assertTrue(all(b['cache_control'] == {'type': 'ephemeral', 'ttl': '1h'} for b in pinned)) + + def test_a_bot_moved_to_an_hour_after_its_summary_keeps_working(self): + # The summary is stored as text and marked as each request is sent, + # so a bot that compacted before its move to an hour (or a daemon + # that marked it at five minutes) sends the next turn whole. + client = Client(self.binary, self.path / 'state.sqlite', self.url, + tools='echo,shell', provider='anthropic', family='anthropic', + model='synthetic-claude', key_env='ANTHROPIC_TEST_KEY', + env={**clean_env(), 'ANTHROPIC_TEST_KEY': 'synthetic-anthropic-key'}, + settings={'context_bytes': 8192, 'compact_at': 50}) + self.addCleanup(client.close) + self.assertIn('result', client.request('create', bot='Bob', workspace=str(self.path), + compaction_instructions='Summarize.', effort='low')) + for n in range(7): + turn = client.request('submit', bot='Bob', request_id=str(n), prompt=f'{n}:' + 'x' * 500)['result']['turn'] + self.assertEqual(client.finished(turn)['data']['status'], 'completed') + self.assertIsNotNone(client.request('inspect', bot='Bob')['result']['compaction']) + self.assertEqual(client.request('configure', bot='Bob', cache_ttl='1h')['result']['changed'], ['cache_ttl']) + self.model.cache_control = {'type': 'ephemeral', 'ttl': '1h'} + while not self.model.requests.empty(): + self.model.requests.get() + turn = client.request('submit', bot='Bob', request_id='moved', prompt='after the move')['result']['turn'] + self.assertEqual(client.finished(turn)['data']['status'], 'completed') + call = self.model.requests.get(timeout=1) + summary = call['messages'][0]['content'][0] + self.assertTrue(summary['text'].startswith('[compaction summary')) + self.assertEqual(summary['cache_control'], {'type': 'ephemeral', 'ttl': '1h'}) + def test_a_summary_inside_a_turn_copies_the_call_before_it_whole(self): # Anthropic reads a cache where a breakpoint went, and the call put # its own at its end, so a summary inside a turn copies that call diff --git a/tests/test_runtime.py b/tests/test_runtime.py index a1611f229..15c61fa29 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -621,6 +621,21 @@ def do_POST(self): for block in request.get('system', []): assert block['text'] and block['cache_control'] == cache assert request['cache_control'] == cache + # The API reads breakpoints in order (tools, system, messages, + # then the request's own last one) and refuses a longer lifetime + # after a shorter one. + marks = [b['cache_control'] for b in request.get('tools', []) + request.get('system', []) + + [b for m in request['messages'] for b in m['content']] if 'cache_control' in b] + ttls = [{'5m': 300, '1h': 3600}[m.get('ttl', '5m')] for m in marks + [request['cache_control']]] + if any(later > earlier for earlier, later in zip(ttls, ttls[1:])): + body = (b'{"type":"error","error":{"type":"invalid_request_error","message":' + b'"a ttl=1h cache_control block must not come after a ttl=5m cache_control block"}}') + self.send_response(400) + self.send_header('Content-Length', str(len(body))) + self.end_headers() + self.wfile.write(body) + self.wfile.flush() + return summary = is_summary(request) # A summary of its own disables tool calls; a copy of the bot's # call keeps its tool choice, which the message cache covers.