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.