diff --git a/.repository-projection.json b/.repository-projection.json index f75342c0b..50c005b54 100644 --- a/.repository-projection.json +++ b/.repository-projection.json @@ -3,11 +3,11 @@ "projection": "deixic-code", "projectionSchemaVersion": 1, "sourceRepository": "dx-corp/mono", - "sourceSha": "633d6832088197307c20f213f1f223f43dbd45eb", + "sourceSha": "664f8600292aa090326e1b67047b1115626e95ee", "destinationRepository": "dx-corp/code", - "priorProjectedBase": "033a0950721198d7bf5ca8004e988482ce3bf234", + "priorProjectedBase": "24a04d23600ea24373ff726583e6bdffc9192d71", "definitionDigest": "82936441c776e3e8edb5d215a75007ec9714a233f489d460075d79d5ef5ba32f", "toolDigest": "89cbdbe1d79917bad52655817aab0eb545b89183915380eaedc3217f014e6715", - "contentDigest": "5c2d1b66ab844ec7b443e56334c005ef4719e4c3447c0775c46f9ef42d1efb67", + "contentDigest": "c6790f993bba397f0d933aef5b096727bf3d2508be70cdfbcb44494093f5c98a", "publicationEligible": true } diff --git a/packages/ai-rs/src/anthropic.rs b/packages/ai-rs/src/anthropic.rs index fce7ff638..23076e558 100644 --- a/packages/ai-rs/src/anthropic.rs +++ b/packages/ai-rs/src/anthropic.rs @@ -369,16 +369,61 @@ impl AnthropicClient { ) -> Result { let model = provider_model_name(&config.model); let capabilities = anthropic_request_capabilities(Some("anthropic"), &model); - let previous_checkpoint = config + let finalized = config .cache_topology .as_ref() - .and_then(|prepared| prepared.previous_checkpoint()) - .filter(|index| *index < messages.len()) - .and_then(|index| { - transform_messages_for_target(&messages[..=index], OutboundTarget::Anthropic) - .len() - .checked_sub(1) - }); + .is_some_and(|prepared| prepared.boundary().is_final()); + // New boundary plans are opt-in. Off keeps the markers main already sent. + let apply_plan = finalized && config.explicit_cache_boundaries; + let plan = config + .cache_topology + .as_ref() + .filter(|prepared| apply_plan && prepared.boundary().is_final()) + .map(|prepared| prepared.boundary()); + let mark_system = if apply_plan { + plan.is_some_and(|plan| plan.mark_system) + } else { + config.cache_system_prompt + }; + let mark_tools = if apply_plan { + plan.is_some_and(|plan| plan.mark_tools) + } else { + config.cache_system_prompt + }; + let previous_checkpoint = if apply_plan { + None + } else { + config + .cache_topology + .as_ref() + .and_then(|prepared| prepared.previous_checkpoint()) + .filter(|index| *index < messages.len()) + .and_then(|index| { + crate::cache_topology::wire_message_index( + messages, + index, + OutboundTarget::Anthropic, + ) + }) + }; + let history_indexes: Vec = if apply_plan { + plan.map(|plan| { + plan.history_indexes + .iter() + .filter_map(|index| { + crate::cache_topology::wire_message_index( + messages, + *index, + OutboundTarget::Anthropic, + ) + }) + .collect() + }) + .unwrap_or_default() + } else { + Vec::new() + }; + let history_ttl = plan.and_then(|plan| plan.history_ttl).unwrap_or("5m"); let messages = transform_messages_for_target(messages, OutboundTarget::Anthropic); let mut body = serde_json::json!({ "model": model, @@ -389,7 +434,7 @@ impl AnthropicClient { // Add system prompt with optional caching if let Some(system) = &config.system { - if config.cache_system_prompt { + if mark_system { // Use cache_control format for prompt caching body["system"] = serde_json::json!([{ "type": "text", @@ -411,7 +456,7 @@ impl AnthropicClient { // Add tools with cache_control on the last tool for tool definition caching if !config.tools.is_empty() { - if config.cache_system_prompt { + if mark_tools { // Add cache_control to the last tool for tool definition caching let mut tools_json: Vec = config.tools.iter().map(|t| serde_json::json!(t)).collect(); @@ -468,7 +513,13 @@ impl AnthropicClient { } if let Some(prepared) = &config.cache_topology { - if config.cache_system_prompt { + if apply_plan { + crate::cache_topology::mark_history_indexes( + &mut body, + history_ttl, + &history_indexes, + ); + } else if config.cache_system_prompt { crate::cache_topology::mark_stable_history(&mut body, "5m", previous_checkpoint); } prepared.append_volatile_tail(&mut body); @@ -1194,6 +1245,301 @@ data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text assert_eq!(headers.get("x-api-key").unwrap(), "test-key"); } + #[test] + fn stable_boundary_wire_contract_marks_anthropic_prefix_before_the_volatile_tail() { + let client = AnthropicClient::new("test-key").unwrap(); + let history = vec![ + Message { + role: Role::User, + content: MessageContent::text("a"), + }, + Message { + role: Role::User, + content: MessageContent::text("b"), + }, + ]; + let mut config = RequestConfig { + model: "claude-sonnet-4-5".into(), + max_tokens: 128, + system: Some("Standing policy".into()), + cache_system_prompt: true, + explicit_cache_boundaries: true, + ..Default::default() + }; + let first = crate::cache_topology::PreparedPrompt::prepare( + &history, + &config, + "session".into(), + None, + ) + .unwrap(); + let appended = vec![ + Message { + role: Role::User, + content: MessageContent::text("a"), + }, + Message { + role: Role::User, + content: MessageContent::text("b"), + }, + Message { + role: Role::Assistant, + content: MessageContent::text("c"), + }, + Message { + role: Role::User, + content: MessageContent::text("d"), + }, + ]; + let mut next = crate::cache_topology::PreparedPrompt::prepare( + &appended, + &config, + "session".into(), + Some(first.topology()), + ) + .unwrap() + .with_volatile_tail(Some("clock".into())); + next.finalize_boundary( + Some("anthropic"), + "claude-sonnet-4-5", + true, + true, + false, + appended.len(), + ) + .unwrap(); + assert_eq!( + next.topology().transition, + crate::cache_topology::CacheTransition::Append + ); + config.cache_topology = Some(next); + let body = client.build_request_body(&appended, &config).unwrap(); + assert_eq!(body["system"][0]["text"], "Standing policy"); + assert_eq!(body["system"][0]["cache_control"]["type"], "ephemeral"); + assert!(body["system"][0]["cache_control"].get("ttl").is_none()); + let messages = body["messages"].as_array().unwrap(); + let marked: Vec = messages + .iter() + .enumerate() + .filter(|(_, message)| { + message["content"].as_array().is_some_and(|blocks| { + blocks + .iter() + .any(|block| block.get("cache_control").is_some()) + }) + }) + .map(|(index, _)| index) + .collect(); + assert_eq!(marked, vec![1, 3]); + assert_eq!(messages.last().unwrap()["content"], "clock"); + assert!(messages.last().unwrap().get("cache_control").is_none()); + assert!(!body.to_string().contains("prompt_cache_breakpoint")); + let markers = body.to_string().matches("cache_control").count(); + assert!( + markers <= 4, + "anthropic allows four breakpoints, saw {markers}" + ); + } + + #[test] + fn explicit_plan_marks_the_checkpoint_after_transform_drops_a_message() { + let client = AnthropicClient::new("test-key").unwrap(); + let history = vec![ + Message { + role: Role::Assistant, + content: MessageContent::Blocks(vec![ContentBlock::Thinking { + thinking: " ".into(), + signature: None, + }]), + }, + Message { + role: Role::User, + content: MessageContent::text("checkpoint"), + }, + Message { + role: Role::Assistant, + content: MessageContent::Blocks(vec![ContentBlock::ToolUse { + id: "tool-1".into(), + name: "read".into(), + input: serde_json::json!({}), + gemini_context: None, + }]), + }, + ]; + let mut config = RequestConfig { + model: "claude-sonnet-4-5".into(), + max_tokens: 128, + system: Some("Standing policy".into()), + cache_system_prompt: true, + explicit_cache_boundaries: true, + ..Default::default() + }; + let prefix = crate::cache_topology::PreparedPrompt::prepare( + &history[..2], + &config, + "session".into(), + None, + ) + .unwrap(); + let mut prepared = crate::cache_topology::PreparedPrompt::prepare( + &history, + &config, + "session".into(), + Some(prefix.topology()), + ) + .unwrap(); + prepared + .finalize_boundary( + Some("anthropic"), + "claude-sonnet-4-5", + true, + true, + false, + history.len(), + ) + .unwrap(); + // The empty thinking message is dropped, so canonical index 1 is wire index 0. + assert_eq!( + crate::cache_topology::wire_message_index( + &history, + 1, + crate::transform::OutboundTarget::Anthropic + ), + Some(0) + ); + config.cache_topology = Some(prepared); + let body = client.build_request_body(&history, &config).unwrap(); + let messages = body["messages"].as_array().unwrap(); + let marked_text: Vec = messages + .iter() + .filter_map(|message| { + let blocks = message["content"].as_array()?; + let marked = blocks + .iter() + .any(|block| block.get("cache_control").is_some()); + if !marked { + return None; + } + blocks.iter().find_map(|block| { + block + .get("text") + .and_then(|text| text.as_str()) + .map(str::to_owned) + }) + }) + .collect(); + assert!( + marked_text.iter().any(|text| text == "checkpoint"), + "marker landed on {marked_text:?}, messages={messages:?}" + ); + assert!(body["system"][0].get("cache_control").is_some()); + } + + #[test] + fn history_rewrite_still_writes_system_and_tools() { + let client = AnthropicClient::new("test-key").unwrap(); + let tools = vec![Tool::new("read", "Read a file")]; + let first_history = vec![Message { + role: Role::User, + content: MessageContent::text("old"), + }]; + let mut config = RequestConfig { + model: "claude-sonnet-4-5".into(), + max_tokens: 128, + system: Some("Standing policy".into()), + tools: tools.into(), + cache_system_prompt: true, + explicit_cache_boundaries: true, + ..Default::default() + }; + let first = crate::cache_topology::PreparedPrompt::prepare( + &first_history, + &config, + "session".into(), + None, + ) + .unwrap(); + let rewritten = vec![Message { + role: Role::User, + content: MessageContent::text("summary"), + }]; + let mut prepared = crate::cache_topology::PreparedPrompt::prepare( + &rewritten, + &config, + "session".into(), + Some(first.topology()), + ) + .unwrap(); + assert_eq!( + prepared.topology().transition, + crate::cache_topology::CacheTransition::HistoryRewritten + ); + prepared + .finalize_boundary( + Some("anthropic"), + "claude-sonnet-4-5", + true, + true, + true, + rewritten.len(), + ) + .unwrap(); + config.cache_topology = Some(prepared); + let body = client.build_request_body(&rewritten, &config).unwrap(); + assert!(body["system"][0].get("cache_control").is_some()); + let tools = body["tools"].as_array().unwrap(); + assert!(tools.last().unwrap().get("cache_control").is_some()); + } + + #[test] + fn explicit_boundaries_off_matches_the_legacy_anthropic_payload() { + let client = AnthropicClient::new("test-key").unwrap(); + let history = vec![ + Message { + role: Role::User, + content: MessageContent::text("a"), + }, + Message { + role: Role::User, + content: MessageContent::text("b"), + }, + ]; + let mut legacy = RequestConfig { + model: "claude-sonnet-4-5".into(), + max_tokens: 128, + system: Some("Standing policy".into()), + cache_system_prompt: true, + ..Default::default() + }; + legacy.cache_topology = Some( + crate::cache_topology::PreparedPrompt::prepare( + &history, + &legacy, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("clock".into())), + ); + let mut opted_out = legacy.clone(); + let mut prepared = opted_out.cache_topology.take().unwrap(); + prepared + .finalize_boundary( + Some("anthropic"), + "claude-sonnet-4-5", + true, + true, + false, + history.len(), + ) + .unwrap(); + opted_out.explicit_cache_boundaries = false; + opted_out.cache_topology = Some(prepared); + let legacy_body = client.build_request_body(&history, &legacy).unwrap(); + let off_body = client.build_request_body(&history, &opted_out).unwrap(); + assert_eq!(legacy_body, off_body); + } + #[test] fn test_build_request_body_with_caching() { let client = AnthropicClient::new("test-key").unwrap(); diff --git a/packages/ai-rs/src/bedrock.rs b/packages/ai-rs/src/bedrock.rs index 222e5e6b6..56e5bd829 100644 --- a/packages/ai-rs/src/bedrock.rs +++ b/packages/ai-rs/src/bedrock.rs @@ -742,6 +742,28 @@ mod tests { let messages = build_request_messages(&history, &config).unwrap(); assert_eq!(cache_point_positions(&messages), vec![1, 3]); + let mut sealed = config.cache_topology.clone().unwrap(); + sealed + .finalize_boundary( + Some("bedrock"), + &config.model, + true, + false, + false, + history.len(), + ) + .unwrap(); + let off = RequestConfig { + cache_topology: Some(sealed), + explicit_cache_boundaries: false, + ..config.clone() + }; + let off_messages = build_request_messages(&history, &off).unwrap(); + assert_eq!( + format!("{off_messages:?}"), + format!("{messages:?}"), + "explicit boundaries stay off, so the Bedrock payload matches the legacy payload" + ); assert_eq!(messages.len(), 5); assert_eq!( messages[4].content(), diff --git a/packages/ai-rs/src/cache_topology.rs b/packages/ai-rs/src/cache_topology.rs index 7bb34397a..8698fb8f7 100644 --- a/packages/ai-rs/src/cache_topology.rs +++ b/packages/ai-rs/src/cache_topology.rs @@ -4,6 +4,40 @@ use anyhow::{Result, ensure}; pub use maestro_runtime_contracts::cache_topology::{CacheTopology, CacheTransition}; use maestro_runtime_contracts::cache_topology::{PromptShape, digest}; +/// Markers chosen before sealing. Indexes are canonical history positions, never the volatile tail. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct BoundaryPlan { + pub record_id: &'static str, + pub history_indexes: Vec, + pub mark_system: bool, + pub mark_tools: bool, + pub history_ttl: Option<&'static str>, + /// GPT-5.6+ Responses explicit-only mode. False unless a breakpoint can sit before the volatile tail. + pub openai_explicit_breakpoint: bool, + finalized: bool, +} + +impl Default for BoundaryPlan { + fn default() -> Self { + Self { + record_id: "unfinalized", + history_indexes: Vec::new(), + mark_system: false, + mark_tools: false, + history_ttl: None, + openai_explicit_breakpoint: false, + finalized: false, + } + } +} + +impl BoundaryPlan { + #[must_use] + pub fn is_final(&self) -> bool { + self.finalized + } +} + #[derive(Clone, Debug)] pub struct PreparedPrompt { topology: CacheTopology, @@ -16,6 +50,11 @@ pub struct PreparedPrompt { /// second marker at this index pins the prefix the previous request /// already paid to write. previous_checkpoint: Option, + /// Compatible earlier checkpoint for this same route, including after A→B→A. + /// This does not change `topology.transition`. + prior_history_index: Option, + prior_tools_system: bool, + boundary: BoundaryPlan, } impl PreparedPrompt { @@ -43,6 +82,9 @@ impl PreparedPrompt { affinity, volatile_tail: None, previous_checkpoint, + prior_history_index: None, + prior_tools_system: false, + boundary: BoundaryPlan::default(), }) } /// Standalone summaries carry no explicit cache hints and never advance the primary checkpoint. @@ -63,9 +105,83 @@ impl PreparedPrompt { affinity: None, volatile_tail: None, previous_checkpoint: None, + prior_history_index: None, + prior_tools_system: false, + boundary: BoundaryPlan::default(), }) } + /// Remember a same-route checkpoint without relabeling the provenance transition. + /// A history rewrite keeps a tools/system prefix only when that prefix was materialized + /// and its digests still match. A different instruction or a reordered history does not match. + pub fn note_prior_route( + &mut self, + prior: &CacheTopology, + tools_system_materialized: bool, + messages: &[Message], + ) -> Result<()> { + ensure!( + !self.boundary.finalized, + "sealed cache boundary cannot change" + ); + if prior.shape.namespace != self.topology.shape.namespace + || prior.shape.model != self.topology.shape.model + { + return Ok(()); + } + let same_tools = prior.shape.tools == self.topology.shape.tools; + let same_instructions = prior.shape.instructions == self.topology.shape.instructions; + if tools_system_materialized && same_tools && same_instructions { + self.prior_tools_system = true; + } + let current: Vec = messages.iter().map(digest).collect(); + if same_tools + && same_instructions + && prior.shape.thinking == self.topology.shape.thinking + && !prior.shape.history.is_empty() + && prior.shape.history.len() < current.len() + && current.starts_with(&prior.shape.history) + { + let index = prior.shape.history.len() - 1; + if self + .prior_history_index + .is_none_or(|existing| index < existing) + { + self.prior_history_index = Some(index); + } + } + Ok(()) + } + + /// Fix marker policy from the sourced capability record. A second call is rejected. + pub fn finalize_boundary( + &mut self, + provider: Option<&str>, + model: &str, + cache_system_prompt: bool, + has_system: bool, + has_tools: bool, + messages_len: usize, + ) -> Result<()> { + ensure!( + !self.boundary.finalized, + "sealed cache boundary cannot change" + ); + self.boundary = plan_boundaries(BoundaryInputs { + provider, + model, + topology: &self.topology, + cache_system_prompt, + has_system, + has_tools, + messages_len, + has_volatile_tail: self.volatile_tail.is_some(), + previous_checkpoint: self.previous_checkpoint, + prior_history_index: self.prior_history_index, + }); + Ok(()) + } + /// Index of the message that closed the previous primary request, when the /// current request appends to that history. pub fn previous_checkpoint(&self) -> Option { @@ -75,10 +191,18 @@ impl PreparedPrompt { /// newer clock, plan, listing, voice setting, or custom instruction. It is /// serialized after history, outside the reusable prefix identity. pub fn with_volatile_tail(mut self, tail: Option) -> Self { + if self.boundary.finalized { + return self; + } self.volatile_tail = tail.filter(|tail| !tail.trim().is_empty()); self } + #[must_use] + pub fn boundary(&self) -> &BoundaryPlan { + &self.boundary + } + pub fn volatile_tail(&self) -> Option<&str> { self.volatile_tail.as_deref() } @@ -158,6 +282,65 @@ pub(crate) fn messages_with_volatile_tail<'a>( std::borrow::Cow::Owned(request) } +/// `MAESTRO_EXPLICIT_CACHE_BOUNDARIES=1` opts into new explicit cache-write +/// boundaries. Unset or any other value keeps today's request bytes. +#[must_use] +pub fn explicit_cache_boundaries_enabled() -> bool { + std::env::var("MAESTRO_EXPLICIT_CACHE_BOUNDARIES") + .ok() + .is_some_and(|value| { + matches!( + value.trim().to_ascii_lowercase().as_str(), + "1" | "true" | "yes" | "on" + ) + }) +} + +/// Canonical history index after `transform_messages_for_target` drops empty +/// messages and inserts synthetic tool results. This is the same translation +/// the previous Anthropic checkpoint used. +pub(crate) fn wire_message_index( + messages: &[crate::Message], + canonical: usize, + target: crate::transform::OutboundTarget, +) -> Option { + if canonical >= messages.len() { + return None; + } + crate::transform::transform_messages_for_target(&messages[..=canonical], target) + .len() + .checked_sub(1) +} + +pub(crate) fn mark_history_indexes(body: &mut serde_json::Value, ttl: &str, indexes: &[usize]) { + let marker = serde_json::json!({"type":"ephemeral", "ttl":ttl}); + let Some(messages) = body + .get_mut("messages") + .and_then(serde_json::Value::as_array_mut) + else { + return; + }; + let mut planned: Vec = indexes + .iter() + .copied() + .filter(|index| *index < messages.len()) + .collect(); + planned.sort_unstable(); + planned.dedup(); + for (nth, index) in planned.iter().copied().enumerate() { + let floor = if nth == 0 { + 0 + } else { + planned[nth - 1].saturating_add(1) + }; + for message in messages[floor..=index].iter_mut().rev() { + if mark_message(message, &marker) { + break; + } + } + } +} + /// Shape each primary checkpoint, including the first request after compaction. /// Call before appending the volatile tail and before attesting provider bytes. /// @@ -212,6 +395,126 @@ fn mark_message(message: &mut serde_json::Value, marker: &serde_json::Value) -> false } +fn take_marker(slots: &mut usize, indexes: &mut Vec, index: usize, messages_len: usize) { + if *slots == 0 || indexes.contains(&index) || index >= messages_len { + return; + } + indexes.push(index); + *slots -= 1; +} + +struct BoundaryInputs<'a> { + provider: Option<&'a str>, + model: &'a str, + topology: &'a CacheTopology, + cache_system_prompt: bool, + has_system: bool, + has_tools: bool, + messages_len: usize, + has_volatile_tail: bool, + previous_checkpoint: Option, + prior_history_index: Option, +} + +fn plan_boundaries(input: BoundaryInputs<'_>) -> BoundaryPlan { + let BoundaryInputs { + provider, + model, + topology, + cache_system_prompt, + has_system, + has_tools, + messages_len, + has_volatile_tail, + previous_checkpoint, + prior_history_index, + } = input; + use crate::{ + CacheBehavior, CacheBoundaryKind, allows_openai_explicit_breakpoint, cache_capability, + }; + if topology.transition == CacheTransition::Auxiliary { + return BoundaryPlan { + record_id: "auxiliary", + finalized: true, + ..BoundaryPlan::default() + }; + } + let capability = cache_capability(provider, model); + let newest = messages_len.checked_sub(1); + let anthropic_explicit = cache_system_prompt + && capability.behavior == CacheBehavior::Explicit + && capability + .boundary_kinds + .contains(&CacheBoundaryKind::AnthropicCacheControl); + // A history rewrite invalidates old history checkpoints. It does not stop + // this request from writing the current tools/system prefix. Eligibility of + // an earlier prefix is recorded on the checkpoint, not a reason to omit the marker. + let mark_system = anthropic_explicit && has_system; + let mark_tools = anthropic_explicit && has_tools; + let mut slots = usize::from(capability.max_explicit_markers.unwrap_or(0)); + if mark_system { + slots = slots.saturating_sub(1); + } + if mark_tools { + slots = slots.saturating_sub(1); + } + let mut history_indexes = Vec::new(); + if anthropic_explicit { + if let Some(index) = newest { + take_marker(&mut slots, &mut history_indexes, index, messages_len); + } + if topology.transition != CacheTransition::HistoryRewritten { + if let Some(index) = previous_checkpoint { + take_marker(&mut slots, &mut history_indexes, index, messages_len); + } + } + // A→B→A may reuse A's checkpoint. Append already recorded that index as + // previous_checkpoint, and a rewrite must not pretend the old history matches. + if !matches!( + topology.transition, + CacheTransition::HistoryRewritten | CacheTransition::Append + ) { + if let Some(index) = prior_history_index { + take_marker(&mut slots, &mut history_indexes, index, messages_len); + } + } + } + let openai_explicit = + has_volatile_tail && allows_openai_explicit_breakpoint(provider, model) && newest.is_some(); + if openai_explicit { + // Not a documented provider maximum. It only stops one request from + // asking for an unbounded number of breakpoints. + const OPENAI_PLANNER_BREAKPOINT_BUDGET: u8 = 4; + slots = usize::from( + capability + .max_explicit_markers + .unwrap_or(OPENAI_PLANNER_BREAKPOINT_BUDGET), + ); + history_indexes.clear(); + if let Some(index) = newest { + take_marker(&mut slots, &mut history_indexes, index, messages_len); + } + if let Some(index) = previous_checkpoint { + take_marker(&mut slots, &mut history_indexes, index, messages_len); + } + if topology.transition != CacheTransition::Append { + if let Some(index) = prior_history_index { + take_marker(&mut slots, &mut history_indexes, index, messages_len); + } + } + } + history_indexes.sort_unstable(); + BoundaryPlan { + record_id: capability.record_id, + history_indexes, + mark_system, + mark_tools, + history_ttl: anthropic_explicit.then_some("5m"), + openai_explicit_breakpoint: openai_explicit, + finalized: true, + } +} + pub(crate) fn validate_prepared(messages: &[Message], config: &RequestConfig) -> Result<()> { if let Some(prepared) = &config.cache_topology { prepared.validate(messages, config)?; @@ -405,5 +708,259 @@ mod tests { PreparedPrompt::prepare(&messages, &config, "local".into(), Some(primary.topology())) .unwrap(); assert_eq!(next.topology().generation, primary.topology().generation); + assert_ne!( + auxiliary.topology().transition, + CacheTransition::Append, + "an auxiliary summary must not take the primary transition" + ); + } + + fn finalize( + prepared: &mut PreparedPrompt, + provider: &str, + model: &str, + config: &RequestConfig, + len: usize, + ) { + prepared + .finalize_boundary( + Some(provider), + model, + config.cache_system_prompt, + config.system.is_some(), + !config.tools.is_empty(), + len, + ) + .unwrap(); + } + + #[test] + fn prefix_dependency_rejects_reordered_or_reinstructed_history() { + let mut config = RequestConfig { + cache_system_prompt: true, + system: Some("instruction-a".into()), + ..Default::default() + }; + let first_messages = vec![user("document")]; + let first = + PreparedPrompt::prepare(&first_messages, &config, "session".into(), None).unwrap(); + config.system = Some("instruction-b".into()); + let mut second = PreparedPrompt::prepare( + &first_messages, + &config, + "session".into(), + Some(first.topology()), + ) + .unwrap(); + second + .note_prior_route(first.topology(), true, &first_messages) + .unwrap(); + assert_eq!(second.prior_history_index, None); + assert!(!second.prior_tools_system); + + config.system = Some("instruction-a".into()); + let reordered = vec![user("document"), user("preface")]; + // Identical document text after a different preceding message is not the same prefix. + let prefix = PreparedPrompt::prepare(&reordered, &config, "session".into(), None).unwrap(); + let swapped = vec![user("preface"), user("document"), user("extra")]; + let mut other = + PreparedPrompt::prepare(&swapped, &config, "session".into(), Some(prefix.topology())) + .unwrap(); + other + .note_prior_route(prefix.topology(), true, &swapped) + .unwrap(); + assert_eq!( + other.prior_history_index, None, + "a longer history with a different leading message is not the same prefix" + ); + let extended = vec![user("document"), user("preface"), user("extra")]; + let mut compatible = PreparedPrompt::prepare( + &extended, + &config, + "session".into(), + Some(prefix.topology()), + ) + .unwrap(); + compatible + .note_prior_route(prefix.topology(), true, &extended) + .unwrap(); + assert_eq!(compatible.prior_history_index, Some(1)); + } + + #[test] + fn volatile_tail_preserves_an_explicit_stable_boundary() { + let config = RequestConfig { + model: "gpt-5.6".into(), + ..Default::default() + }; + let history = vec![user("stable")]; + let mut first = PreparedPrompt::prepare(&history, &config, "session".into(), None) + .unwrap() + .with_volatile_tail(Some("clock: 1".into())); + finalize(&mut first, "openai", "gpt-5.6", &config, history.len()); + let mut next = + PreparedPrompt::prepare(&history, &config, "session".into(), Some(first.topology())) + .unwrap() + .with_volatile_tail(Some("clock: 2".into())); + finalize(&mut next, "openai", "gpt-5.6", &config, history.len()); + assert_eq!(next.topology().generation, first.topology().generation); + assert_eq!(next.topology().transition, CacheTransition::Append); + assert_eq!(next.boundary().history_indexes, vec![0]); + assert!(next.boundary().openai_explicit_breakpoint); + assert!(!next.boundary().history_indexes.contains(&1)); + } + + #[test] + fn compaction_keeps_only_a_materialized_tools_prefix() { + let config = RequestConfig { + cache_system_prompt: true, + system: Some("standing".into()), + tools: std::sync::Arc::new(vec![crate::Tool::new("read", "Read")]), + ..Default::default() + }; + let history = vec![user("a"), user("b")]; + let first = PreparedPrompt::prepare(&history, &config, "session".into(), None).unwrap(); + let rewritten = vec![user("summary"), user("b")]; + let mut after = PreparedPrompt::prepare( + &rewritten, + &config, + "session".into(), + Some(first.topology()), + ) + .unwrap(); + assert_eq!( + after.topology().transition, + CacheTransition::HistoryRewritten + ); + after + .note_prior_route(first.topology(), false, &rewritten) + .unwrap(); + finalize( + &mut after, + "anthropic", + "claude-sonnet-4-5", + &config, + rewritten.len(), + ); + assert_eq!(after.boundary().history_indexes, vec![rewritten.len() - 1]); + assert!(after.boundary().mark_system && after.boundary().mark_tools); + assert!(!after.prior_tools_system); + assert_eq!(after.prior_history_index, None); + assert_eq!( + after.topology().transition, + CacheTransition::HistoryRewritten + ); + + let mut carried = PreparedPrompt::prepare( + &rewritten, + &config, + "session".into(), + Some(first.topology()), + ) + .unwrap(); + carried + .note_prior_route(first.topology(), true, &rewritten) + .unwrap(); + finalize( + &mut carried, + "anthropic", + "claude-sonnet-4-5", + &config, + rewritten.len(), + ); + assert!(carried.boundary().mark_system); + assert!(carried.boundary().mark_tools); + assert_eq!(carried.prior_history_index, None); + assert_eq!( + carried.topology().transition, + CacheTransition::HistoryRewritten + ); + } + + #[test] + fn returning_route_checkpoint_is_not_relabeled_append() { + let config = RequestConfig::default(); + let route_a = vec![user("a1"), user("a2")]; + let mut model_a = config.clone(); + model_a.model = "route-a".into(); + model_a.cache_system_prompt = true; + let mut first = + PreparedPrompt::prepare(&route_a, &model_a, "session".into(), None).unwrap(); + finalize( + &mut first, + "anthropic", + "claude-sonnet-4-5", + &model_a, + route_a.len(), + ); + let mut model_b = model_a.clone(); + model_b.model = "route-b".into(); + let on_b = + PreparedPrompt::prepare(&route_a, &model_b, "session".into(), Some(first.topology())) + .unwrap(); + assert_eq!(on_b.topology().transition, CacheTransition::ModelChanged); + let continued = vec![user("a1"), user("a2"), user("a3")]; + let mut back = PreparedPrompt::prepare( + &continued, + &model_a, + "session".into(), + Some(on_b.topology()), + ) + .unwrap(); + assert_eq!(back.topology().transition, CacheTransition::ModelChanged); + back.note_prior_route(first.topology(), true, &continued) + .unwrap(); + assert_eq!(back.prior_history_index, Some(1)); + finalize( + &mut back, + "anthropic", + "claude-sonnet-4-5", + &model_a, + continued.len(), + ); + assert_eq!(back.topology().transition, CacheTransition::ModelChanged); + assert!(back.boundary().history_indexes.contains(&1)); + assert!(back.boundary().history_indexes.contains(&2)); + assert_eq!(back.boundary().history_indexes.len(), 2); + } + + #[test] + fn mutation_of_a_sealed_boundary_fails() { + let mut config = RequestConfig { + model: "gpt-5.6".into(), + ..Default::default() + }; + let messages = vec![user("stable")]; + let mut prepared = PreparedPrompt::prepare(&messages, &config, "session".into(), None) + .unwrap() + .with_volatile_tail(Some("clock".into())); + finalize(&mut prepared, "openai", "gpt-5.6", &config, messages.len()); + let prior = prepared.topology().clone(); + assert!(prepared.note_prior_route(&prior, true, &messages).is_err()); + assert!( + prepared + .finalize_boundary(Some("openai"), "gpt-5.6", true, true, true, 1) + .is_err() + ); + let tail = prepared.volatile_tail().unwrap().to_string(); + let sealed = prepared.with_volatile_tail(Some("replaced".into())); + assert_eq!(sealed.volatile_tail(), Some(tail.as_str())); + config.system = Some("changed policy".into()); + assert!(sealed.validate(&messages, &config).is_err()); + } + + #[test] + fn different_namespace_does_not_inherit_a_checkpoint() { + let messages = vec![user("same")]; + let config = RequestConfig::default(); + let tenant_a = + PreparedPrompt::prepare(&messages, &config, "tenant-a".into(), None).unwrap(); + let mut tenant_b = + PreparedPrompt::prepare(&messages, &config, "tenant-b".into(), None).unwrap(); + tenant_b + .note_prior_route(tenant_a.topology(), true, &messages) + .unwrap(); + assert_eq!(tenant_b.prior_history_index, None); + assert!(tenant_a.validate_namespace("tenant-b").is_err()); } } diff --git a/packages/ai-rs/src/lib.rs b/packages/ai-rs/src/lib.rs index 2c48a860e..5849ef5fd 100644 --- a/packages/ai-rs/src/lib.rs +++ b/packages/ai-rs/src/lib.rs @@ -88,8 +88,11 @@ mod kimi; pub mod managed_authorization; mod model_capabilities; pub use model_capabilities::{ - ASTRA_CONTEXT_TOKENS, ASTRA_OUTPUT_TOKENS, AnthropicRequestCapabilities, AnthropicThinkingMode, - OpenAiRequestCapabilities, OpenAiWireProtocol, anthropic_request_capabilities, + ANTHROPIC_PROMPT_CACHING_SOURCE, ASTRA_CONTEXT_TOKENS, ASTRA_OUTPUT_TOKENS, + AnthropicRequestCapabilities, AnthropicThinkingMode, CacheBehavior, CacheBoundaryKind, + CacheCapability, CacheLookback, CacheRetention, CacheUsageInterpretation, CacheWireProtocol, + OPENAI_PROMPT_CACHING_SOURCE, OpenAiRequestCapabilities, OpenAiWireProtocol, + allows_openai_explicit_breakpoint, anthropic_request_capabilities, cache_capability, openai_request_capabilities, supports_explicit_prompt_caching, }; pub mod op_secret; diff --git a/packages/ai-rs/src/model_capabilities.rs b/packages/ai-rs/src/model_capabilities.rs index 3cc089a6b..bba893235 100644 --- a/packages/ai-rs/src/model_capabilities.rs +++ b/packages/ai-rs/src/model_capabilities.rs @@ -536,6 +536,395 @@ impl OpenAiRequestCapabilities { } } +/// OpenAI prompt-caching guide. Re-check before adding a wire field. +pub const OPENAI_PROMPT_CACHING_SOURCE: &str = + "https://developers.openai.com/api/docs/guides/prompt-caching (checked 2026-09-24)"; +/// Anthropic prompt-caching guide. Re-check before adding a wire field. +pub const ANTHROPIC_PROMPT_CACHING_SOURCE: &str = + "https://platform.claude.com/docs/en/build-with-claude/prompt-caching (checked 2026-09-24)"; + +/// How a route caches. Unknown does not inherit another route's lifetime or rates. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CacheBehavior { + Unknown, + Unsupported, + Implicit, + Explicit, +} + +/// Where a caller is allowed to place a boundary. Empty means "do not send a marker". +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CacheBoundaryKind { + /// Anthropic `cache_control` on tools, system, or a non-thinking message block. + AnthropicCacheControl, + /// OpenAI Responses `prompt_cache_breakpoint` on a content block. Not valid on top-level `instructions`. + OpenAiContentBreakpoint, + /// The provider chooses breakpoints. Callers must not send a marker parameter. + ProviderChosen, +} + +/// How far a later request can see a breakpoint that was actually written. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CacheLookback { + Unknown, + /// Anthropic checks at most this many blocks per breakpoint, counting the breakpoint. + Blocks(u32), + /// OpenAI explicit lookup: the first N and the latest M explicit breakpoints. + ExplicitBreakpoints { + first: u32, + latest: u32, + }, +} + +/// Retention the source actually states. A missing hint is not five minutes. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CacheRetention { + Unknown, + Unsupported, + /// Anthropic default ephemeral lifetime. The clock starts when the caching request starts. + AnthropicEphemeral5m, + /// Anthropic `ttl: "1h"`. Twice the base input price. Not selected by the boundary planner. + AnthropicEphemeral1h, + /// GPT-5.6+ `prompt_cache_options.ttl` of `"30m"`, also the default. Minimum lifetime after write or reuse. + OpenAiTtl30m, + /// Earlier OpenAI models: `in_memory` is typically 5–10 minutes of inactivity, up to an hour. + OpenAiInMemory, + /// Earlier OpenAI `prompt_cache_retention: "24h"`. Up to 24 hours. Not a single cutoff. + OpenAiExtended24h, + /// Earlier OpenAI default depends on the organization's zero-data-retention policy, which this process does not know. + OpenAiOrganizationDependent, +} + +impl CacheRetention { + /// A single sourced duration, in seconds, suitable only as a diagnostic hint. + /// `None` means the source does not give one number. Callers must not substitute 300. + #[must_use] + pub fn hint_seconds(self) -> Option { + match self { + Self::AnthropicEphemeral5m => Some(300), + Self::AnthropicEphemeral1h => Some(3_600), + Self::OpenAiTtl30m => Some(1_800), + Self::Unknown + | Self::Unsupported + | Self::OpenAiInMemory + | Self::OpenAiExtended24h + | Self::OpenAiOrganizationDependent => None, + } + } +} + +/// How reported usage relates to cached and uncached input. Missing stays missing. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CacheUsageInterpretation { + Unknown, + /// `input_tokens` excludes cache read and cache creation. Total input is the sum of the three. + AnthropicSeparateCacheFields, + /// GPT-5.6+: `input_tokens` includes `cached_tokens` and `cache_write_tokens`. No 128-token rounding. + OpenAiInclusiveExact, + /// Earlier OpenAI: `cached_tokens` omits hidden system tokens and rounds down to a multiple of 128. + OpenAiCachedTokensRounded, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CacheWireProtocol { + Unknown, + None, + AnthropicMessages, + BedrockConverse, + OpenAiResponses, + OpenAiChat, +} + +/// Model-and-provider scoped cache contract. Rates are sourced milli-units of the +/// uncached input price (100 = 0.1×). `None` is not a default of 100 or of 1.25×. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CacheCapability { + pub behavior: CacheBehavior, + pub boundary_kinds: &'static [CacheBoundaryKind], + pub minimum_cacheable_tokens: Option, + pub max_explicit_markers: Option, + pub lookback: CacheLookback, + pub read_rate_millis: Option, + pub write_rate_millis: Option, + pub retention: CacheRetention, + pub wire: CacheWireProtocol, + pub usage: CacheUsageInterpretation, + /// Attribution for this record. Not a provider semver. + pub record_id: &'static str, + pub source: &'static str, +} + +const NO_BOUNDARIES: &[CacheBoundaryKind] = &[]; +const ANTHROPIC_BOUNDARIES: &[CacheBoundaryKind] = &[CacheBoundaryKind::AnthropicCacheControl]; +const OPENAI_EXPLICIT_BOUNDARIES: &[CacheBoundaryKind] = + &[CacheBoundaryKind::OpenAiContentBreakpoint]; +const PROVIDER_CHOSEN_BOUNDARIES: &[CacheBoundaryKind] = &[CacheBoundaryKind::ProviderChosen]; + +/// Family, minimum cacheable tokens, cache-read milli-rate. Longer families come first +/// so `claude-opus-4` does not swallow `claude-opus-4-5`. +const ANTHROPIC_CACHE_TABLE: &[(&str, u32, u32)] = &[ + ("claude-fable-5-1", 512, 25), + ("claude-mythos-5-1", 512, 25), + ("claude-opus-5-5", 512, 50), + ("claude-opus-5", 512, 100), + ("claude-fable-5", 512, 100), + ("claude-mythos-preview", 2_048, 100), + ("claude-mythos-5", 512, 100), + ("claude-opus-4-8", 1_024, 100), + ("claude-opus-4-7", 2_048, 100), + ("claude-opus-4-6", 4_096, 100), + ("claude-opus-4-5", 4_096, 100), + ("claude-sonnet-5", 1_024, 100), + ("claude-sonnet-4-6", 1_024, 100), + ("claude-sonnet-4-5", 1_024, 100), + ("claude-haiku-4-5", 4_096, 100), + ("claude-haiku-3-5", 2_048, 100), +]; + +fn unknown_capability(record_id: &'static str, source: &'static str) -> CacheCapability { + CacheCapability { + behavior: CacheBehavior::Unknown, + boundary_kinds: NO_BOUNDARIES, + minimum_cacheable_tokens: None, + max_explicit_markers: None, + lookback: CacheLookback::Unknown, + read_rate_millis: None, + write_rate_millis: None, + retention: CacheRetention::Unknown, + wire: CacheWireProtocol::Unknown, + usage: CacheUsageInterpretation::Unknown, + record_id, + source, + } +} + +fn normalized_route(provider: Option<&str>, model: &str) -> (Option, String) { + let stripped = strip_managed_model_prefix(model.trim()); + let embedded = stripped + .split_once('/') + .map(|(name, _)| name.trim().to_ascii_lowercase()); + let provider = provider + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(|name| name.to_ascii_lowercase()) + .or(embedded); + (provider, provider_model_name(stripped).to_ascii_lowercase()) +} + +fn openai_explicit_breakpoint_family(model: &str) -> bool { + // GPT-5.6 and later, per the OpenAI guide's model table. GPT-5.5 and earlier are implicit only. + let name = model.rsplit('/').next().unwrap_or(model); + let Some(rest) = name.strip_prefix("gpt-") else { + return false; + }; + let major_end = rest + .find(|c: char| !c.is_ascii_digit()) + .unwrap_or(rest.len()); + if major_end == 0 { + return false; + } + let Ok(major) = rest[..major_end].parse::() else { + return false; + }; + if major > 5 { + return true; + } + if major < 5 { + return false; + } + let after = &rest[major_end..]; + let Some(minor_src) = after.strip_prefix('.') else { + return false; + }; + let minor_end = minor_src + .find(|c: char| !c.is_ascii_digit()) + .unwrap_or(minor_src.len()); + minor_src[..minor_end] + .parse::() + .ok() + .is_some_and(|minor| minor >= 6) +} + +fn anthropic_cache_row(model: &str) -> Option<(u32, u32)> { + let normalized = anthropic_model_id(Some("anthropic"), model).or_else(|| { + let lower = provider_model_name(model).to_ascii_lowercase(); + lower.starts_with("claude-").then_some(lower) + })?; + ANTHROPIC_CACHE_TABLE + .iter() + .find(|(family, _, _)| is_model_family(&normalized, family)) + .map(|(_, minimum, read)| (*minimum, *read)) +} + +fn anthropic_explicit_capability( + model: &str, + wire: CacheWireProtocol, + record_id: &'static str, + known_retention: bool, +) -> CacheCapability { + let row = anthropic_cache_row(model); + CacheCapability { + behavior: CacheBehavior::Explicit, + boundary_kinds: ANTHROPIC_BOUNDARIES, + minimum_cacheable_tokens: row.map(|(minimum, _)| minimum), + max_explicit_markers: Some(4), + lookback: CacheLookback::Blocks(20), + // Recognized families use the published multiplier. An unrecognized Claude id + // stays explicit (the route already accepts markers) but does not inherit 0.1×. + read_rate_millis: row.map(|(_, read)| read), + write_rate_millis: row.map(|_| 1_250), + retention: if known_retention { + CacheRetention::AnthropicEphemeral5m + } else { + CacheRetention::Unknown + }, + wire, + usage: if wire == CacheWireProtocol::AnthropicMessages { + CacheUsageInterpretation::AnthropicSeparateCacheFields + } else { + // Bedrock's usage field names are a different document and were not re-fetched here. + CacheUsageInterpretation::Unknown + }, + record_id, + source: ANTHROPIC_PROMPT_CACHING_SOURCE, + } +} + +/// Sourced cache contract for this provider route and model. Aggregator routes stay +/// unknown even when the model id looks like a direct OpenAI or Anthropic model. +#[must_use] +pub fn cache_capability(provider: Option<&str>, model: &str) -> CacheCapability { + // A managed or aggregator id does not identify the physical deployment. + // Do not inherit the direct OpenAI or Anthropic contract from the suffix. + if has_managed_model_prefix(model) { + return unknown_capability( + "managed-gateway-route.2026-09-24", + "managed gateway routes do not identify the physical cache", + ); + } + let (provider_name, model_id) = normalized_route(provider, model); + let provider = provider_name.as_deref(); + let is = |name: &str| provider.is_some_and(|provider| provider == name); + + if is("openrouter") || is("azure") || is("azure-openai") || is("vertex-ai") || is("google") { + return unknown_capability( + "aggregator-or-translated-route.2026-09-24", + "physical cache placement is not the logical model id", + ); + } + if is("anthropic") + || is("claude") + || (provider.is_none() && anthropic_model_id(None, model).is_some()) + { + return anthropic_explicit_capability( + model, + CacheWireProtocol::AnthropicMessages, + "anthropic-messages-explicit.2026-09-24", + true, + ); + } + if is("bedrock") { + return if supports_explicit_prompt_caching(AiProvider::Bedrock, model) { + anthropic_explicit_capability( + &model_id, + CacheWireProtocol::BedrockConverse, + "bedrock-claude-explicit-markers.2026-09-24", + false, + ) + } else { + unknown_capability( + "bedrock-unlisted.2026-09-24", + ANTHROPIC_PROMPT_CACHING_SOURCE, + ) + }; + } + if is("openai") { + return openai_direct_capability(provider, model); + } + if is("llamacpp") { + return CacheCapability { + behavior: CacheBehavior::Implicit, + boundary_kinds: PROVIDER_CHOSEN_BOUNDARIES, + minimum_cacheable_tokens: None, + max_explicit_markers: Some(0), + lookback: CacheLookback::Unknown, + read_rate_millis: None, + write_rate_millis: None, + retention: CacheRetention::Unknown, + wire: CacheWireProtocol::OpenAiChat, + usage: CacheUsageInterpretation::Unknown, + record_id: "llamacpp-cache-prompt.2026-09-24", + source: "existing llama.cpp cache_prompt request flag; no sourced lifetime", + }; + } + unknown_capability( + "unknown-route.2026-09-24", + "no sourced cache contract for this provider", + ) +} + +fn openai_direct_capability(provider: Option<&str>, model: &str) -> CacheCapability { + let responses = uses_responses_api(provider, model); + if responses && openai_explicit_breakpoint_family(model) { + return CacheCapability { + behavior: CacheBehavior::Explicit, + boundary_kinds: OPENAI_EXPLICIT_BOUNDARIES, + minimum_cacheable_tokens: Some(1_024), + // The guide documents a lookup window (first 2 and latest 50), not a + // hard marker cap. The planner applies its own budget separately. + max_explicit_markers: None, + lookback: CacheLookback::ExplicitBreakpoints { + first: 2, + latest: 50, + }, + read_rate_millis: Some(100), + write_rate_millis: Some(1_250), + retention: CacheRetention::OpenAiTtl30m, + wire: CacheWireProtocol::OpenAiResponses, + usage: CacheUsageInterpretation::OpenAiInclusiveExact, + record_id: "openai-responses-gpt-5.6-explicit.2026-09-24", + source: OPENAI_PROMPT_CACHING_SOURCE, + }; + } + CacheCapability { + behavior: CacheBehavior::Implicit, + boundary_kinds: PROVIDER_CHOSEN_BOUNDARIES, + minimum_cacheable_tokens: None, + max_explicit_markers: Some(0), + lookback: CacheLookback::Unknown, + // The guide says earlier models have a model-dependent cached-input rate and no + // extra write charge. Neither fact is a universal 0.1× read rate. + read_rate_millis: None, + write_rate_millis: Some(1_000), + retention: CacheRetention::OpenAiOrganizationDependent, + wire: if responses { + CacheWireProtocol::OpenAiResponses + } else { + CacheWireProtocol::OpenAiChat + }, + usage: CacheUsageInterpretation::OpenAiCachedTokensRounded, + record_id: "openai-implicit.2026-09-24", + source: OPENAI_PROMPT_CACHING_SOURCE, + } +} + +#[must_use] +pub fn allows_openai_explicit_breakpoint(provider: Option<&str>, model: &str) -> bool { + let capability = cache_capability(provider, model); + capability.behavior == CacheBehavior::Explicit + && capability.wire == CacheWireProtocol::OpenAiResponses + && capability + .boundary_kinds + .contains(&CacheBoundaryKind::OpenAiContentBreakpoint) +} + #[cfg(test)] mod tests { use super::*; @@ -574,6 +963,109 @@ mod tests { } } + #[test] + fn cache_capability_records_are_sourced_and_do_not_lend_defaults() { + let unknown = cache_capability(Some("openrouter"), "openai/gpt-5.6"); + assert_eq!(unknown.behavior, CacheBehavior::Unknown); + assert!(unknown.boundary_kinds.is_empty()); + assert_eq!(unknown.read_rate_millis, None); + assert_eq!(unknown.write_rate_millis, None); + assert_eq!(unknown.retention.hint_seconds(), None); + assert!(!allows_openai_explicit_breakpoint( + Some("openrouter"), + "openai/gpt-5.6" + )); + + let bare = cache_capability(None, "gpt-5.6"); + assert_ne!(bare.behavior, CacheBehavior::Explicit); + assert_eq!(bare.read_rate_millis, None); + assert_eq!(bare.retention.hint_seconds(), None); + + let explicit = cache_capability(Some("openai"), "gpt-5.6-sol"); + assert_eq!(explicit.behavior, CacheBehavior::Explicit); + assert_eq!(explicit.minimum_cacheable_tokens, Some(1_024)); + assert_eq!(explicit.max_explicit_markers, None); + assert_eq!( + explicit.lookback, + CacheLookback::ExplicitBreakpoints { + first: 2, + latest: 50 + } + ); + assert_eq!(explicit.read_rate_millis, Some(100)); + assert_eq!(explicit.write_rate_millis, Some(1_250)); + assert_eq!(explicit.retention, CacheRetention::OpenAiTtl30m); + assert_eq!( + explicit.usage, + CacheUsageInterpretation::OpenAiInclusiveExact + ); + assert!(explicit.source.contains("developers.openai.com")); + assert!(allows_openai_explicit_breakpoint( + Some("openai"), + "gpt-6-astra" + )); + assert!( + !allows_openai_explicit_breakpoint(Some("openai"), "maestro-managed/openai/gpt-5.6"), + "a managed-gateway model id is not a confirmed physical OpenAI route" + ); + assert_eq!( + cache_capability(Some("openai"), "claude-opus-4").minimum_cacheable_tokens, + None + ); + assert_eq!( + cache_capability(Some("anthropic"), "claude-sonnet-4").minimum_cacheable_tokens, + None, + "families absent from the sourced minimum table do not inherit a neighbor's number" + ); + assert_eq!( + cache_capability(Some("anthropic"), "claude-opus-4-1").read_rate_millis, + None + ); + + let earlier = cache_capability(Some("openai"), "gpt-5.4"); + assert_eq!(earlier.behavior, CacheBehavior::Implicit); + assert_eq!(earlier.max_explicit_markers, Some(0)); + assert_eq!(earlier.read_rate_millis, None); + assert_eq!(earlier.write_rate_millis, Some(1_000)); + assert_eq!(earlier.retention.hint_seconds(), None); + assert!(!allows_openai_explicit_breakpoint(Some("openai"), "gpt-5")); + assert!(!allows_openai_explicit_breakpoint( + Some("openai"), + "gpt-4.1" + )); + + let opus = cache_capability(Some("anthropic"), "claude-opus-5-5"); + assert_eq!(opus.minimum_cacheable_tokens, Some(512)); + assert_eq!(opus.read_rate_millis, Some(50)); + assert_eq!(opus.lookback, CacheLookback::Blocks(20)); + assert_eq!(opus.max_explicit_markers, Some(4)); + assert_eq!(opus.retention.hint_seconds(), Some(300)); + assert_eq!( + cache_capability(Some("anthropic"), "claude-fable-5-1").read_rate_millis, + Some(25) + ); + assert_eq!( + cache_capability(Some("anthropic"), "claude-haiku-4-5").minimum_cacheable_tokens, + Some(4_096) + ); + let unrecognized = cache_capability(Some("anthropic"), "claude-not-a-family"); + assert_eq!(unrecognized.behavior, CacheBehavior::Explicit); + assert_eq!(unrecognized.read_rate_millis, None); + assert_eq!(unrecognized.minimum_cacheable_tokens, None); + + let bedrock = cache_capability( + Some("bedrock"), + "bedrock/anthropic.claude-sonnet-4-5-20250929-v1:0", + ); + assert_eq!(bedrock.behavior, CacheBehavior::Explicit); + assert_eq!(bedrock.usage, CacheUsageInterpretation::Unknown); + assert_eq!(bedrock.retention.hint_seconds(), None); + assert_eq!( + cache_capability(Some("bedrock"), "amazon.nova-pro-v1:0").behavior, + CacheBehavior::Unknown + ); + } + #[test] fn maximum_effort_is_provider_and_model_specific() { for (model, expected) in [ diff --git a/packages/ai-rs/src/openai.rs b/packages/ai-rs/src/openai.rs index cb1e77058..ffe7e8c69 100644 --- a/packages/ai-rs/src/openai.rs +++ b/packages/ai-rs/src/openai.rs @@ -1933,14 +1933,19 @@ impl OpenAiClient { config: &RequestConfig, ) -> Result { let model = self.request_model_name(&config.model); + let canonical_messages = messages; let messages = transform_messages_for_target(messages, OutboundTarget::OpenAiResponses); // Convert messages to Responses API format // The input array contains ResponseItems, which can be messages, function calls, or function outputs let mut input: Vec = Vec::new(); + let mut message_end: Vec> = Vec::new(); for msg in messages { + let before = input.len(); match msg.role { - Role::System => continue, // System goes in instructions + Role::System => { + // Instructions are sent separately and cannot hold a breakpoint. + } Role::User => { // User messages use "input_text" content type match &msg.content { @@ -2073,6 +2078,7 @@ impl OpenAiClient { } } } + message_end.push(input.len().checked_sub(1).filter(|_| input.len() > before)); } let mut body = serde_json::json!({ @@ -2141,6 +2147,29 @@ impl OpenAiClient { // This enables streaming of reasoning text (only encrypted_content is valid) body["include"] = serde_json::json!(["reasoning.encrypted_content"]); + // Opt-in only. Managed and aggregator routes stay on the implicit shape even + // when the model id looks like a direct GPT-5.6 deployment. + if config.explicit_cache_boundaries && !self.managed_gateway { + if let Some(prepared) = config.cache_topology.as_ref() { + if prepared.boundary().is_final() + && prepared.boundary().openai_explicit_breakpoint + && crate::allows_openai_explicit_breakpoint( + self.route_provider.as_deref(), + &config.model, + ) + && mark_responses_breakpoints( + &mut body, + canonical_messages, + &message_end, + &prepared.boundary().history_indexes, + ) + { + body["prompt_cache_options"] = + serde_json::json!({"mode": "explicit", "ttl": "30m"}); + } + } + } + Ok(body) } @@ -3391,6 +3420,111 @@ struct PromptTokensDetails { cache_write_tokens: Option, } +fn input_carrier(block_type: Option<&str>) -> bool { + matches!( + block_type, + Some("input_text" | "input_image" | "input_file") + ) +} + +fn item_has_input_carrier(item: &serde_json::Value) -> bool { + item.get("content") + .and_then(serde_json::Value::as_array) + .is_some_and(|content| { + content + .iter() + .any(|block| input_carrier(block.get("type").and_then(serde_json::Value::as_str))) + }) +} + +fn mark_input_carrier(item: &mut serde_json::Value) -> bool { + let Some(content) = item + .get_mut("content") + .and_then(serde_json::Value::as_array_mut) + else { + return false; + }; + let Some(block) = content + .iter_mut() + .rev() + .find(|block| input_carrier(block.get("type").and_then(serde_json::Value::as_str))) + else { + return false; + }; + block["prompt_cache_breakpoint"] = serde_json::json!({"mode": "explicit"}); + true +} + +/// Returns whether the newest planned boundary was marked on a documented input carrier. +/// Older indexes may walk back to an earlier carrier. `output_text` and string tool +/// output are not carriers; if the newest boundary is one of those, the body is left implicit. +fn mark_responses_breakpoints( + body: &mut serde_json::Value, + canonical: &[Message], + message_end: &[Option], + indexes: &[usize], +) -> bool { + let Some(input) = body + .get_mut("input") + .and_then(serde_json::Value::as_array_mut) + else { + return false; + }; + let Some(newest) = indexes.iter().copied().max() else { + return false; + }; + let Some(newest_item) = canonical_input_item(canonical, message_end, newest) else { + return false; + }; + if !item_has_input_carrier(&input[newest_item]) { + return false; + } + let mut placed_newest = false; + let mut planned: Vec = indexes.to_vec(); + planned.sort_unstable(); + planned.dedup(); + for (nth, canonical_index) in planned.iter().copied().enumerate() { + let floor = if nth == 0 { + 0 + } else { + canonical_input_item(canonical, message_end, planned[nth - 1]).unwrap_or(0) + }; + let Some(start) = canonical_input_item(canonical, message_end, canonical_index) else { + continue; + }; + let mut item_index = start; + let mut marked = false; + loop { + let markable = item_has_input_carrier(&input[item_index]); + if markable && mark_input_carrier(&mut input[item_index]) { + marked = true; + break; + } + if item_index <= floor { + break; + } + item_index -= 1; + } + if canonical_index == newest { + placed_newest = marked; + } + } + placed_newest +} + +fn canonical_input_item( + canonical: &[Message], + message_end: &[Option], + canonical_index: usize, +) -> Option { + let wire = crate::cache_topology::wire_message_index( + canonical, + canonical_index, + crate::transform::OutboundTarget::OpenAiResponses, + )?; + message_end.get(wire).copied().flatten() +} + fn apply_prompt_cache_key(body: &mut serde_json::Value, provider: Option<&str>, key: Option<&str>) { if let (Some("openrouter"), Some(key)) = ( provider, @@ -3545,6 +3679,284 @@ mod tests { assert!(auxiliary.get("prompt_cache_key").is_none()); } + #[test] + fn stable_boundary_wire_contract_places_openai_breakpoint_before_the_volatile_tail() { + let client = OpenAiClient::new("fixture") + .unwrap() + .with_route_provider("openai"); + let history = vec![Message { + role: Role::User, + content: MessageContent::text("stable history"), + }]; + let mut config = RequestConfig { + model: "gpt-5.6".into(), + system: Some("Standing developer policy".into()), + explicit_cache_boundaries: true, + ..Default::default() + }; + let mut prepared = crate::cache_topology::PreparedPrompt::prepare( + &history, + &config, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("clock and plan".into())); + prepared + .finalize_boundary(Some("openai"), "gpt-5.6", false, true, false, history.len()) + .unwrap(); + config.cache_topology = Some(prepared); + let body = client + .build_request_body_for_api(&history, &config, true) + .unwrap(); + assert_eq!(body["instructions"], "Standing developer policy"); + assert!(body["instructions"].is_string()); + let input = body["input"].as_array().unwrap(); + assert_eq!(input[0]["role"], "user"); + assert_eq!( + input[0]["content"][0]["prompt_cache_breakpoint"]["mode"], + "explicit" + ); + assert_eq!(input[1]["role"], "user"); + assert_eq!(input[1]["content"], "clock and plan"); + assert!(input[1].get("prompt_cache_breakpoint").is_none()); + assert_eq!(body["prompt_cache_options"]["mode"], "explicit"); + assert_eq!(body["prompt_cache_options"]["ttl"], "30m"); + assert!(body.get("prompt_cache_breakpoint").is_none()); + let rendered = body.to_string(); + assert!(!rendered.contains("\"role\":\"developer\"")); + assert_eq!(rendered.matches("Standing developer policy").count(), 1); + + let mut earlier = config.clone(); + earlier.model = "gpt-5.4".into(); + let mut prepared = crate::cache_topology::PreparedPrompt::prepare( + &history, + &earlier, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("clock and plan".into())); + prepared + .finalize_boundary(Some("openai"), "gpt-5.4", false, true, false, history.len()) + .unwrap(); + earlier.cache_topology = Some(prepared); + let implicit = client + .build_request_body_for_api(&history, &earlier, true) + .unwrap(); + assert!(implicit.get("prompt_cache_options").is_none()); + assert!(!implicit.to_string().contains("prompt_cache_breakpoint")); + + let chat = client + .build_request_body_for_api(&history, &config, false) + .unwrap(); + assert!( + !chat.to_string().contains("prompt_cache_breakpoint"), + "chat completions must not receive the Responses breakpoint" + ); + + let router = OpenAiClient::new("fixture") + .unwrap() + .with_route_provider("openrouter"); + let routed = router + .build_request_body_for_api(&history, &config, true) + .unwrap(); + assert!( + !routed.to_string().contains("prompt_cache_breakpoint"), + "an aggregator route must not receive a guessed OpenAI breakpoint" + ); + + let mut instructions_only = config.clone(); + let mut prepared = crate::cache_topology::PreparedPrompt::prepare( + &[], + &instructions_only, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("only the tail".into())); + prepared + .finalize_boundary(Some("openai"), "gpt-5.6", false, true, false, 0) + .unwrap(); + assert!(!prepared.boundary().openai_explicit_breakpoint); + instructions_only.cache_topology = Some(prepared); + let no_history = client + .build_request_body_for_api(&[], &instructions_only, true) + .unwrap(); + assert_eq!(no_history["instructions"], "Standing developer policy"); + assert!(!no_history.to_string().contains("prompt_cache_breakpoint")); + assert!(no_history.get("prompt_cache_options").is_none()); + } + + #[test] + fn explicit_breakpoint_follows_the_transformed_message_and_refuses_unmarkable_tails() { + let client = OpenAiClient::new("fixture") + .unwrap() + .with_route_provider("openai"); + let history = vec![ + Message { + role: Role::System, + content: MessageContent::text("not on the input array"), + }, + Message { + role: Role::User, + content: MessageContent::text("stable"), + }, + Message { + role: Role::Assistant, + content: MessageContent::text("assistant text"), + }, + Message { + role: Role::User, + content: MessageContent::text("q"), + }, + ]; + let mut config = RequestConfig { + model: "gpt-5.6".into(), + explicit_cache_boundaries: true, + ..Default::default() + }; + let mut prepared = crate::cache_topology::PreparedPrompt::prepare( + &history, + &config, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("clock".into())); + prepared + .finalize_boundary(Some("openai"), "gpt-5.6", true, false, false, history.len()) + .unwrap(); + config.cache_topology = Some(prepared); + let body = client + .build_request_body_for_api(&history, &config, true) + .unwrap(); + let marked: Vec = body["input"] + .as_array() + .unwrap() + .iter() + .filter_map(|item| { + let blocks = item.get("content")?.as_array()?; + let marked = blocks + .iter() + .any(|block| block.get("prompt_cache_breakpoint").is_some()); + if !marked { + return None; + } + blocks + .iter() + .find_map(|block| block.get("text").and_then(|text| text.as_str())) + .map(str::to_owned) + }) + .collect(); + assert_eq!(marked, vec!["q".to_string()]); + let breakpoint_types: Vec<&str> = body["input"] + .as_array() + .unwrap() + .iter() + .flat_map(|item| { + item.get("content") + .and_then(|c| c.as_array()) + .into_iter() + .flatten() + }) + .filter(|block| block.get("prompt_cache_breakpoint").is_some()) + .filter_map(|block| block.get("type").and_then(|t| t.as_str())) + .collect(); + assert_eq!(breakpoint_types, vec!["input_text"]); + + let tool_history = vec![ + Message { + role: Role::User, + content: MessageContent::text("ask"), + }, + Message { + role: Role::Assistant, + content: MessageContent::Blocks(vec![crate::ContentBlock::ToolUse { + id: "call-1".into(), + name: "read".into(), + input: serde_json::json!({}), + gemini_context: None, + }]), + }, + Message { + role: Role::User, + content: MessageContent::Blocks(vec![crate::ContentBlock::ToolResult { + tool_use_id: "call-1".into(), + content: "result".into(), + is_error: None, + }]), + }, + ]; + let mut tool_config = config.clone(); + let mut tool_prepared = crate::cache_topology::PreparedPrompt::prepare( + &tool_history, + &tool_config, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("clock".into())); + tool_prepared + .finalize_boundary( + Some("openai"), + "gpt-5.6", + true, + false, + false, + tool_history.len(), + ) + .unwrap(); + tool_config.cache_topology = Some(tool_prepared); + let tool_body = client + .build_request_body_for_api(&tool_history, &tool_config, true) + .unwrap(); + assert!( + tool_body.get("prompt_cache_options").is_none(), + "a newest tool-result boundary is not an input carrier, so the request stays implicit" + ); + assert!(!tool_body.to_string().contains("prompt_cache_breakpoint")); + + let mut off = config.clone(); + off.explicit_cache_boundaries = false; + let off_body = client + .build_request_body_for_api(&history, &off, true) + .unwrap(); + let mut legacy = off.clone(); + let unfinalized = crate::cache_topology::PreparedPrompt::prepare( + &history, + &legacy, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("clock".into())); + legacy.cache_topology = Some(unfinalized); + let legacy_body = client + .build_request_body_for_api(&history, &legacy, true) + .unwrap(); + assert_eq!(off_body, legacy_body); + + let managed = OpenAiClient::new("fixture") + .unwrap() + .with_route_provider("openai") + .with_managed_gateway_scope( + "org_123", + "workspace_456", + serde_json::json!({ + "provider": "openai", + "environment": "production", + "credential_name": "default" + }), + ) + .unwrap(); + let managed_body = managed + .build_request_body_for_api(&history, &config, true) + .unwrap(); + assert!(!managed_body.to_string().contains("prompt_cache_breakpoint")); + assert!(managed_body.get("prompt_cache_options").is_none()); + } + #[test] fn prompt_cache_affinity_is_opt_in_and_openrouter_only() { let mut body = serde_json::json!({}); diff --git a/packages/ai-rs/src/types.rs b/packages/ai-rs/src/types.rs index 7ec000b20..0a3601e95 100644 --- a/packages/ai-rs/src/types.rs +++ b/packages/ai-rs/src/types.rs @@ -278,6 +278,9 @@ pub struct RequestConfig { /// Enable provider prompt caching for stable system, tool, and history prefixes. /// When true, supported providers add their native cache markers. pub cache_system_prompt: bool, + /// Opt in to new explicit cache-write boundaries. Default off, so outgoing + /// payloads stay identical to the previous request shape. + pub explicit_cache_boundaries: bool, /// Immutable preparation proof; dispatch rejects changes after preparation. pub cache_topology: Option, } @@ -292,6 +295,7 @@ impl Default for RequestConfig { tools: Arc::new(Vec::new()), thinking: None, cache_system_prompt: false, + explicit_cache_boundaries: false, cache_topology: None, } } diff --git a/packages/context-rs/src/token_counting.rs b/packages/context-rs/src/token_counting.rs index fa5c7431c..fc0c2425c 100644 --- a/packages/context-rs/src/token_counting.rs +++ b/packages/context-rs/src/token_counting.rs @@ -50,19 +50,40 @@ pub struct CacheIdentity<'a> { #[serde(rename_all = "snake_case")] pub enum CacheReuse { Reusable, + /// Prepared identity matches and no sourced retention hint applies. + CompatiblePrefix, + /// Compatible, and the gap is inside a sourced retention hint. Not an observed hit. + PredictedReuse, + /// Provider-reported cache read. Not a prediction. + ObservedRead, + /// Provider-reported cache write. Lifetime starts at the provider event. + ObservedWrite, ModelChanged, SystemPromptChanged, ThinkingChanged, SkillsChanged, ToolsChanged, + /// The gap is past a sourced retention hint. Not proof the entry is gone. LikelyExpired, + Unsupported, + /// No sourced lifetime or read rate. Callers must not assume five minutes or 0.1×. + Unknown, } impl CacheReuse { pub fn explanation(self) -> &'static str { match self { - Self::Reusable => { - "Model, system prompt, thinking, and tools match; cache reuse is not confirmed." + Self::Reusable | Self::CompatiblePrefix => { + "Compatible prefix: model, instructions, thinking, and tools match. This is not an observed read or a predicted hit." + } + Self::PredictedReuse => { + "Predicted reuse: the prefix is compatible and the gap is inside the provider retention hint. A preparation timestamp is not a provider cache-creation time." + } + Self::ObservedRead => { + "Observed read: the provider reported cache-read tokens for this prefix." + } + Self::ObservedWrite => { + "Observed write: the provider reported cache-write tokens. The lifetime starts at that event, not at preparation or stream completion." } Self::ModelChanged => "Model changed.", Self::SystemPromptChanged => "System prompt or instructions changed.", @@ -70,12 +91,34 @@ impl CacheReuse { Self::SkillsChanged => "Skills changed.", Self::ToolsChanged => "Tool schemas changed.", Self::LikelyExpired => { - "At least five minutes since the previous request; cache may have expired." + "Expired hint: the gap since preparation, or since the observed cache event when one was recorded, is past the provider retention hint. This does not confirm the entry is gone, and requesting a quote does not extend it." } + Self::Unsupported => "Unsupported: this route has no prompt-cache markers.", + Self::Unknown => "Unknown cache behavior. No lifetime or read rate is assumed.", } } } +/// Provider-reported cache usage. `None` is missing, not zero. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct CacheObservation { + pub read_tokens: Option, + pub write_tokens: Option, + /// Provider event time. Not preparation time and not stream completion. + /// For Anthropic this is request start: the ephemeral TTL clock starts there. + /// Production audit still calls [`RequestCacheSnapshot::compare`], which passes + /// no observation. Wiring reported usage into this field is deferred. + pub event_seconds: Option, +} + +/// One earlier route's prepared topology. Digests only; no prompt text. +#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +pub struct PriorRouteRecord { + pub topology: maestro_ai::cache_topology::CacheTopology, + #[serde(default)] + pub tools_system_materialized: bool, +} + /// Diagnostic request identity, persisted by the existing session owner. /// This predicts reuse; it never substitutes for provider-reported cache usage. #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] @@ -87,6 +130,21 @@ pub struct RequestCacheSnapshot { pub prepared_at_seconds: u64, #[serde(default, skip_serializing_if = "Option::is_none")] pub cache_topology: Option, + /// Routed provider id. Absent on legacy snapshots; inference then uses the model id only. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub provider: Option, + /// The tools/system breakpoint was placed on this request, not merely eligible. + #[serde(default)] + pub tools_system_materialized: bool, + /// Other routes' prepared topologies for this session. Empty on legacy snapshots. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub prior_routes: Vec, + /// Capability record that chose the markers. Absent on legacy snapshots. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub boundary_record_id: Option, + /// Canonical history indexes the plan marked. Empty when the plan was not finalized. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub history_boundaries: Vec, } impl RequestCacheSnapshot { @@ -111,31 +169,103 @@ impl RequestCacheSnapshot { .cache_topology .as_ref() .map(|prepared| prepared.topology().clone()), + provider: None, + tools_system_materialized: config.cache_topology.as_ref().is_some_and(|prepared| { + let plan = prepared.boundary(); + plan.is_final() && (plan.mark_system || plan.mark_tools) + }), + prior_routes: Vec::new(), + boundary_record_id: config.cache_topology.as_ref().and_then(|prepared| { + prepared + .boundary() + .is_final() + .then(|| prepared.boundary().record_id.to_string()) + }), + history_boundaries: config + .cache_topology + .as_ref() + .filter(|prepared| prepared.boundary().is_final()) + .map(|prepared| prepared.boundary().history_indexes.clone()) + .unwrap_or_default(), } } + /// Keep other routes when the model changes. Same-model topology stays on `cache_topology`. + /// At most two prior routes: this is the boundary handoff, not a placement registry. + #[must_use] + pub fn rolled_prior_routes(&self, next_model_digest: &str) -> Vec { + let mut kept = Vec::new(); + if let Some(topology) = &self.cache_topology { + if topology.shape.model != next_model_digest { + kept.push(PriorRouteRecord { + topology: topology.clone(), + tools_system_materialized: self.tools_system_materialized, + }); + } + } + for prior in &self.prior_routes { + if prior.topology.shape.model != next_model_digest + && kept + .iter() + .all(|existing| existing.topology.shape.model != prior.topology.shape.model) + { + kept.push(prior.clone()); + } + } + kept.truncate(2); + kept + } + pub fn compare(&self, previous: &Self) -> CacheReuse { - fn identity(value: &RequestCacheSnapshot) -> CacheIdentity<'_> { - CacheIdentity { - model: &value.model, - system_prompt_sha256: &value.system_sha256, - thinking: &value.thinking_sha256, - skills_sha256: "", // Effective skill text is part of the request prompt. + self.compare_observed(previous, None) + } + + /// Diagnostic only. Does not store a new expiry and does not treat preparation as cache creation. + pub fn compare_observed( + &self, + previous: &Self, + observed: Option<&CacheObservation>, + ) -> CacheReuse { + if self.model != previous.model { + return CacheReuse::ModelChanged; + } + if self.system_sha256 != previous.system_sha256 { + return CacheReuse::SystemPromptChanged; + } + if self.thinking_sha256 != previous.thinking_sha256 { + return CacheReuse::ThinkingChanged; + } + if self.tools_sha256 != previous.tools_sha256 { + return CacheReuse::ToolsChanged; + } + let provider = self.provider.as_deref().or(previous.provider.as_deref()); + let capability = maestro_ai::cache_capability(provider, &self.model); + if capability.behavior == maestro_ai::CacheBehavior::Unsupported { + return CacheReuse::Unsupported; + } + if let Some(observed) = observed { + if observed.read_tokens.is_some_and(|tokens| tokens > 0) { + return CacheReuse::ObservedRead; + } + if observed.write_tokens.is_some_and(|tokens| tokens > 0) { + return CacheReuse::ObservedWrite; } } - let reason = cache_reuse( - &identity(previous), - &identity(self), - self.prepared_at_seconds - .saturating_sub(previous.prepared_at_seconds), - 300, - ); - if matches!(reason, CacheReuse::Reusable | CacheReuse::LikelyExpired) - && self.tools_sha256 != previous.tools_sha256 - { - CacheReuse::ToolsChanged + let Some(hint) = capability.retention.hint_seconds() else { + return if capability.behavior == maestro_ai::CacheBehavior::Unknown { + CacheReuse::Unknown + } else { + CacheReuse::CompatiblePrefix + }; + }; + let anchor = observed + .and_then(|observation| observation.event_seconds) + .unwrap_or(previous.prepared_at_seconds); + let idle = self.prepared_at_seconds.saturating_sub(anchor); + if idle >= hint { + CacheReuse::LikelyExpired } else { - reason + CacheReuse::PredictedReuse } } } @@ -240,7 +370,12 @@ mod tests { serde_json::from_str(&serde_json::to_string(&initial).unwrap()).unwrap(); assert_eq!( RequestCacheSnapshot::from_request(&config, 11).compare(&restored), - CacheReuse::Reusable + CacheReuse::PredictedReuse + ); + assert!( + CacheReuse::PredictedReuse + .explanation() + .contains("not a provider cache-creation time") ); assert_eq!( RequestCacheSnapshot::from_request(&config, 311).compare(&restored), @@ -347,4 +482,143 @@ mod tests { CacheReuse::LikelyExpired ); } + + #[test] + fn lifetime_hints_are_provider_specific_and_preparation_is_not_creation() { + let mut config = maestro_ai::RequestConfig { + model: "local-gguf".into(), + ..Default::default() + }; + let previous = RequestCacheSnapshot::from_request(&config, 10); + let later = RequestCacheSnapshot::from_request(&config, 10_000); + assert_eq!(later.compare(&previous), CacheReuse::Unknown); + assert!( + CacheReuse::Unknown + .explanation() + .contains("No lifetime or read rate") + ); + assert_eq!( + maestro_ai::cache_capability(None, "local-gguf").read_rate_millis, + None + ); + assert_eq!( + maestro_ai::cache_capability(None, "local-gguf") + .retention + .hint_seconds(), + None + ); + + config.model = "gpt-5.6".into(); + let mut prepared = RequestCacheSnapshot::from_request(&config, 0); + prepared.provider = Some("openai".into()); + let mut within = RequestCacheSnapshot::from_request(&config, 1_000); + within.provider = Some("openai".into()); + assert_eq!(within.compare(&prepared), CacheReuse::PredictedReuse); + let mut past_hint = RequestCacheSnapshot::from_request(&config, 1_800); + past_hint.provider = Some("openai".into()); + assert_eq!(past_hint.compare(&prepared), CacheReuse::LikelyExpired); + // The same preparation gap is not expired when the provider event is recent. + // Repeating the comparison does not move that event. + let observed = CacheObservation { + read_tokens: None, + write_tokens: Some(100_000), + event_seconds: Some(1_700), + }; + assert_eq!( + past_hint.compare_observed(&prepared, Some(&observed)), + CacheReuse::ObservedWrite + ); + assert_eq!( + past_hint.compare_observed(&prepared, Some(&observed)), + CacheReuse::ObservedWrite + ); + let read = CacheObservation { + read_tokens: Some(100_000), + write_tokens: Some(0), + event_seconds: Some(1_790), + }; + assert_eq!( + past_hint.compare_observed(&prepared, Some(&read)), + CacheReuse::ObservedRead + ); + let explicit_zero = CacheObservation { + read_tokens: Some(0), + write_tokens: Some(0), + event_seconds: Some(1_790), + }; + assert_eq!( + past_hint.compare_observed(&prepared, Some(&explicit_zero)), + CacheReuse::PredictedReuse + ); + let missing = CacheObservation { + read_tokens: None, + write_tokens: None, + event_seconds: Some(0), + }; + assert_eq!( + past_hint.compare_observed(&prepared, Some(&missing)), + CacheReuse::LikelyExpired + ); + + config.model = "gpt-5.4".into(); + let mut earlier = RequestCacheSnapshot::from_request(&config, 0); + earlier.provider = Some("openai".into()); + let mut much_later = RequestCacheSnapshot::from_request(&config, 86_400); + much_later.provider = Some("openai".into()); + assert_eq!(much_later.compare(&earlier), CacheReuse::CompatiblePrefix); + assert_eq!( + much_later.compare(&earlier), + much_later.compare(&earlier), + "a repeated diagnostic must not extend the hint" + ); + } + + #[test] + fn legacy_cache_snapshot_without_capability_fields_stays_readable() { + let legacy = r#"{"model":"claude-sonnet-4-5","system_sha256":"abc","thinking_sha256":"def","tools_sha256":"ghi","prepared_at_seconds":10}"#; + let restored: RequestCacheSnapshot = serde_json::from_str(legacy).unwrap(); + assert!(restored.provider.is_none()); + assert!(!restored.tools_system_materialized); + assert!(restored.prior_routes.is_empty()); + assert!(restored.boundary_record_id.is_none()); + assert!(restored.history_boundaries.is_empty()); + assert!(restored.cache_topology.is_none()); + let mut config = maestro_ai::RequestConfig { + model: "gpt-5.6".into(), + explicit_cache_boundaries: true, + ..Default::default() + }; + let messages = vec![maestro_ai::Message { + role: maestro_ai::Role::User, + content: maestro_ai::MessageContent::text("stable"), + }]; + let mut prepared = maestro_ai::cache_topology::PreparedPrompt::prepare( + &messages, + &config, + "session".into(), + None, + ) + .unwrap() + .with_volatile_tail(Some("clock".into())); + prepared + .finalize_boundary(Some("openai"), "gpt-5.6", true, false, false, 1) + .unwrap(); + config.cache_topology = Some(prepared); + let snapshot = RequestCacheSnapshot::from_request(&config, 10); + let round_trip: RequestCacheSnapshot = + serde_json::from_str(&serde_json::to_string(&snapshot).unwrap()).unwrap(); + assert_eq!( + round_trip.boundary_record_id.as_deref(), + Some("openai-responses-gpt-5.6-explicit.2026-09-24") + ); + assert_eq!(round_trip.history_boundaries, vec![0]); + let topology = r#"{"version":1,"generation":1,"transition":"initial","shape":{"namespace":"n","model":"m","instructions":"i","tools":"t","thinking":"h","cache_policy":"c","history":[]}}"#; + let parsed: maestro_ai::cache_topology::CacheTopology = + serde_json::from_str(topology).unwrap(); + assert_eq!(parsed.generation, 1); + assert_eq!( + parsed.transition, + maestro_ai::cache_topology::CacheTransition::Initial + ); + } } diff --git a/packages/local-host-rs/src/agent/native_admission_tests.rs b/packages/local-host-rs/src/agent/native_admission_tests.rs index 24f642914..5c7f05551 100644 --- a/packages/local-host-rs/src/agent/native_admission_tests.rs +++ b/packages/local-host-rs/src/agent/native_admission_tests.rs @@ -647,6 +647,7 @@ fn test_request_config_building() { tools: tools.into(), thinking: None, cache_system_prompt: true, + explicit_cache_boundaries: false, cache_topology: None, }; @@ -655,6 +656,7 @@ fn test_request_config_building() { assert!(request_config.system.is_some()); assert!(!request_config.tools.is_empty()); assert!(request_config.cache_system_prompt); + assert!(!request_config.explicit_cache_boundaries); } #[test] diff --git a/packages/local-host-rs/src/semantic_text.rs b/packages/local-host-rs/src/semantic_text.rs index b572a3d30..cb654ee35 100644 --- a/packages/local-host-rs/src/semantic_text.rs +++ b/packages/local-host-rs/src/semantic_text.rs @@ -33,10 +33,43 @@ enum Following { Fence { marker: char, width: usize }, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum TitleLex { + Leading, + Hashes(u8), + Heading, + FirstStar, + Bold { last_star: bool }, + BoldClosed { colon: bool }, + Other, +} + +impl TitleLex { + fn advance(self, ch: char) -> Self { + match self { + Self::Leading if ch == ' ' => Self::Leading, + Self::Leading if ch == '#' => Self::Hashes(1), + Self::Leading if ch == '*' => Self::FirstStar, + Self::Hashes(width) if ch == '#' && width < 6 => Self::Hashes(width + 1), + Self::Hashes(_) if ch == ' ' => Self::Heading, + Self::Heading => Self::Heading, + Self::FirstStar if ch == '*' => Self::Bold { last_star: false }, + Self::Bold { last_star: true } if ch == '*' => Self::BoldClosed { colon: false }, + Self::Bold { .. } => Self::Bold { + last_star: ch == '*', + }, + Self::BoldClosed { colon } if ch.is_whitespace() => Self::BoldClosed { colon }, + Self::BoldClosed { colon: false } if ch == ':' => Self::BoldClosed { colon: true }, + _ => Self::Other, + } + } +} + #[derive(Debug, Clone)] pub struct SemanticTextRelease { pending: String, line: String, + title_lex: TitleLex, line_non_title: bool, following: Following, held_since_ms: Option, @@ -57,6 +90,7 @@ impl SemanticTextRelease { Self { pending: String::new(), line: String::new(), + title_lex: TitleLex::Leading, line_non_title: false, following: Following::None, held_since_ms: None, @@ -87,6 +121,7 @@ impl SemanticTextRelease { .then_some(FlushReason::SizeLimit); self.release(now_ms, forced, &mut releases); self.line.clear(); + self.title_lex = TitleLex::Other; self.line_non_title = true; } if self.pending.is_empty() { @@ -96,6 +131,7 @@ impl SemanticTextRelease { self.peak_held_bytes = self.peak_held_bytes.max(self.pending.len()); if !self.line_non_title { self.line.push(ch); + self.title_lex = self.title_lex.advance(ch); } if ch == '\n' { if self.line_non_title { @@ -104,13 +140,15 @@ impl SemanticTextRelease { self.finish_line(now_ms, &mut releases); } self.line.clear(); + self.title_lex = TitleLex::Leading; self.line_non_title = false; } else if self.following == Following::None && !self.line_non_title - && !could_be_title(&self.line) + && self.title_lex == TitleLex::Other { // Ordinary prose need not wait for a line or a transport timer. self.line.clear(); + self.title_lex = TitleLex::Other; self.line_non_title = true; } if self.pending.len() >= self.max_bytes { @@ -118,6 +156,7 @@ impl SemanticTextRelease { .then_some(FlushReason::SizeLimit); self.release(now_ms, forced, &mut releases); self.line.clear(); + self.title_lex = TitleLex::Other; self.line_non_title = true; } } @@ -131,6 +170,7 @@ impl SemanticTextRelease { self.release(now_ms, Some(FlushReason::LatencyLimit), &mut releases); if !self.line.is_empty() { self.line.clear(); + self.title_lex = TitleLex::Other; self.line_non_title = true; } } @@ -174,6 +214,7 @@ impl SemanticTextRelease { self.release(now_ms, Some(reason), &mut releases); self.following = Following::None; self.line.clear(); + self.title_lex = TitleLex::Leading; self.line_non_title = false; releases } @@ -240,24 +281,6 @@ impl SemanticTextRelease { } } -fn could_be_title(line: &str) -> bool { - let value = line.trim_start_matches(' '); - if value.starts_with('#') { - let width = value.bytes().take_while(|byte| *byte == b'#').count(); - return width <= 6 && (value.len() == width || value.as_bytes().get(width) == Some(&b' ')); - } - if value == "*" { - return true; - } - if let Some(body) = value.strip_prefix("**") { - if let Some(end) = body.find("**") { - return body[end + 2..].trim().trim_end_matches(':').is_empty(); - } - return true; - } - false -} - fn standalone_title(line: &str) -> bool { let value = line.trim(); if value.starts_with('#') { @@ -535,4 +558,21 @@ mod tests { assert!(releases.iter().all(|release| release.text.len() <= 4_096)); assert_eq!(policy.held_bytes(), 0); } + + #[test] + fn long_bold_candidate_is_scanned_incrementally_and_forced_at_the_size_limit() { + let mut policy = SemanticTextRelease::new(4_096, 10_000); + assert!(policy.push("**", 0).is_empty()); + let mut released = String::new(); + for _ in 0..4_100 { + for chunk in policy.push("x", 0) { + released.push_str(&chunk.text); + } + assert!(policy.held_bytes() <= 4_096); + } + for chunk in policy.flush(FlushReason::Finalization, 0) { + released.push_str(&chunk.text); + } + assert_eq!(released, format!("**{}", "x".repeat(4_100))); + } } diff --git a/packages/runtime-rs/src/agent/mod.rs b/packages/runtime-rs/src/agent/mod.rs index e9f07a55d..8dda29301 100644 --- a/packages/runtime-rs/src/agent/mod.rs +++ b/packages/runtime-rs/src/agent/mod.rs @@ -95,5 +95,5 @@ pub use workflow_state::{ pub use native::{ codex_native_effect_denial_for_test, deferred_firewall_verdict_for_test, deferred_policy_rejection_event_for_test, invalidate_cache_after_serial_tool_for_test, - rerun_deferred_pre_tool_use_for_test, + rerun_deferred_pre_tool_use_for_test, take_provider_history_vault_passes_for_bench, }; diff --git a/packages/runtime-rs/src/agent/native.rs b/packages/runtime-rs/src/agent/native.rs index bcd6cf556..3dbe67adb 100644 --- a/packages/runtime-rs/src/agent/native.rs +++ b/packages/runtime-rs/src/agent/native.rs @@ -3737,12 +3737,25 @@ fn vault_provider_history_shared( messages: &Arc>, credential_vault: &CredentialVault, ) -> Result>> { + #[cfg(feature = "test-support")] + PROVIDER_HISTORY_VAULT_PASSES.fetch_add(1, std::sync::atomic::Ordering::Relaxed); Ok(Arc::new(vault_provider_history( messages, credential_vault, )?)) } +#[cfg(feature = "test-support")] +static PROVIDER_HISTORY_VAULT_PASSES: std::sync::atomic::AtomicU64 = + std::sync::atomic::AtomicU64::new(0); + +/// Return and reset the number of full history vault passes in this process. +/// This deterministic work counter is only compiled for benchmark builds. +#[cfg(feature = "test-support")] +pub fn take_provider_history_vault_passes_for_bench() -> u64 { + PROVIDER_HISTORY_VAULT_PASSES.swap(0, std::sync::atomic::Ordering::Relaxed) +} + /// The only direct-provider payload constructed by the native turn loop. /// Its fields stay private so another path cannot pass an unexamined history /// or config to a provider by accident. @@ -3755,10 +3768,21 @@ struct ProviderSafeRequest { impl ProviderSafeRequest { fn prepare( messages: &Arc>, - mut config: RequestConfig, + config: RequestConfig, vault: &CredentialVault, ) -> Result { let messages = vault_provider_history_shared(messages, vault)?; + Self::prepare_vaulted(messages, config, vault) + } + + // The provider loop already vaults its projected history before computing + // request usage. Keep that projection and recheck it after config + // preparation, which can discover credentials in system or tool text. + fn prepare_vaulted( + messages: Arc>, + mut config: RequestConfig, + vault: &CredentialVault, + ) -> Result { config.system = config.system.map(|system| vault.vault_in_text(&system)); config.tools = vault_provider_tools(config.tools.as_ref(), vault)?; // Config preparation may discover a credential present in history. diff --git a/packages/runtime-rs/src/agent/native/context.rs b/packages/runtime-rs/src/agent/native/context.rs index 640ff1b12..53b2d83a8 100644 --- a/packages/runtime-rs/src/agent/native/context.rs +++ b/packages/runtime-rs/src/agent/native/context.rs @@ -2,6 +2,22 @@ use super::*; +fn prior_route_notes( + snapshot: Option<&maestro_context::token_counting::RequestCacheSnapshot>, +) -> Vec<(maestro_ai::cache_topology::CacheTopology, bool)> { + let Some(snapshot) = snapshot else { + return Vec::new(); + }; + let mut notes = Vec::new(); + if let Some(topology) = &snapshot.cache_topology { + notes.push((topology.clone(), snapshot.tools_system_materialized)); + } + for prior in &snapshot.prior_routes { + notes.push((prior.topology.clone(), prior.tools_system_materialized)); + } + notes +} + impl NativeAgentRunner { async fn recover_durable_tool_operations(&mut self) { let mut records = match self.hooks.hook_load_tool_operations().await { @@ -564,6 +580,8 @@ impl NativeAgentRunner { thinking, cache_topology: None, cache_system_prompt, + explicit_cache_boundaries: + maestro_ai::cache_topology::explicit_cache_boundaries_enabled(), }; let mut audit = self .runtime_audit @@ -576,21 +594,32 @@ impl NativeAgentRunner { .map(|client| client.cache_namespace()) .transpose()? .unwrap_or_else(|| "local".into()); + let prior_notes = prior_route_notes(audit.request_cache.as_ref()); let previous = audit .request_cache .as_ref() .and_then(|snapshot| snapshot.cache_topology.as_ref()); - config.cache_topology = Some( - maestro_ai::cache_topology::PreparedPrompt::prepare( - request_messages, - &config, - namespace, - previous, - )? - .with_volatile_tail(self.prompt_context.clone()), - ); + let mut prepared = maestro_ai::cache_topology::PreparedPrompt::prepare( + request_messages, + &config, + namespace, + previous, + )?; + for (topology, materialized) in &prior_notes { + prepared.note_prior_route(topology, *materialized, request_messages)?; + } + prepared = prepared.with_volatile_tail(self.prompt_context.clone()); + prepared.finalize_boundary( + self.client.as_ref().map(|client| client.provider_name()), + &config.model, + config.cache_system_prompt, + config.system.is_some(), + !config.tools.is_empty(), + request_messages.len(), + )?; + config.cache_topology = Some(prepared); } - let snapshot = maestro_context::token_counting::RequestCacheSnapshot::from_request( + let mut snapshot = maestro_context::token_counting::RequestCacheSnapshot::from_request( &config, std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -598,6 +627,17 @@ impl NativeAgentRunner { .as_secs(), ); if include_tools { + if let Some(provider) = self.client.as_ref().map(|client| client.provider_name()) { + snapshot.provider = Some(provider.to_string()); + } + if let Some(previous) = audit.request_cache.as_ref() { + let next_model = snapshot + .cache_topology + .as_ref() + .map(|topology| topology.shape.model.as_str()) + .unwrap_or(""); + snapshot.prior_routes = previous.rolled_prior_routes(next_model); + } audit.cache_reuse = audit .request_cache .as_ref() @@ -628,23 +668,45 @@ impl NativeAgentRunner { .runtime_audit .write() .unwrap_or_else(|p| p.into_inner()); + let prior_notes = prior_route_notes(audit.request_cache.as_ref()); let previous = audit .request_cache .as_ref() .and_then(|snapshot| snapshot.cache_topology.as_ref()); - config.cache_topology = Some( - maestro_ai::cache_topology::PreparedPrompt::prepare( - &messages, &config, namespace, previous, - )? - .with_volatile_tail(self.prompt_context.clone()), - ); - let snapshot = maestro_context::token_counting::RequestCacheSnapshot::from_request( + let mut prepared = maestro_ai::cache_topology::PreparedPrompt::prepare( + &messages, &config, namespace, previous, + )?; + for (topology, materialized) in &prior_notes { + prepared.note_prior_route(topology, *materialized, &messages)?; + } + prepared = prepared.with_volatile_tail(self.prompt_context.clone()); + prepared.finalize_boundary( + self.client.as_ref().map(|client| client.provider_name()), + &config.model, + config.cache_system_prompt, + config.system.is_some(), + !config.tools.is_empty(), + messages.len(), + )?; + config.cache_topology = Some(prepared); + let mut snapshot = maestro_context::token_counting::RequestCacheSnapshot::from_request( &config, std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() .as_secs(), ); + if let Some(provider) = self.client.as_ref().map(|client| client.provider_name()) { + snapshot.provider = Some(provider.to_string()); + } + if let Some(previous) = audit.request_cache.as_ref() { + let next_model = snapshot + .cache_topology + .as_ref() + .map(|topology| topology.shape.model.as_str()) + .unwrap_or(""); + snapshot.prior_routes = previous.rolled_prior_routes(next_model); + } audit.cache_reuse = audit .request_cache .as_ref() @@ -762,9 +824,18 @@ impl NativeAgentRunner { .map(|client| client.cache_namespace()) .transpose()? .unwrap_or_else(|| "local".into()); - config.cache_topology = Some(maestro_ai::cache_topology::PreparedPrompt::auxiliary( - messages, &config, namespace, - )?); + let mut prepared = + maestro_ai::cache_topology::PreparedPrompt::auxiliary(messages, &config, namespace)?; + // Auxiliary summaries never take the primary checkpoint or a routing lease. + prepared.finalize_boundary( + self.client.as_ref().map(|client| client.provider_name()), + &config.model, + false, + config.system.is_some(), + !config.tools.is_empty(), + messages.len(), + )?; + config.cache_topology = Some(prepared); Ok(config) } pub(super) async fn run_selective_summary( diff --git a/packages/runtime-rs/src/agent/native/provider_loop.rs b/packages/runtime-rs/src/agent/native/provider_loop.rs index c9371178b..34ca2c1f9 100644 --- a/packages/runtime-rs/src/agent/native/provider_loop.rs +++ b/packages/runtime-rs/src/agent/native/provider_loop.rs @@ -286,8 +286,11 @@ impl NativeAgentRunner { let (config, request_usage) = self .build_config_with_usage(&provider_messages, true) .await?; - let prepared_request = - ProviderSafeRequest::prepare(&provider_messages, config, &self.credential_vault)?; + let prepared_request = ProviderSafeRequest::prepare_vaulted( + provider_messages, + config, + &self.credential_vault, + )?; let provider_messages = &prepared_request.messages; let config = &prepared_request.config; let estimated_input_tokens = request_usage.total(); diff --git a/packages/runtime-rs/src/agent/native/tests.rs b/packages/runtime-rs/src/agent/native/tests.rs index ad521d6a1..5dfe14214 100644 --- a/packages/runtime-rs/src/agent/native/tests.rs +++ b/packages/runtime-rs/src/agent/native/tests.rs @@ -7332,6 +7332,33 @@ fn provider_request_rechecks_system_after_tool_credential_discovery() { assert!(system.contains("{{CRED|")); } +#[test] +fn provider_request_pre_vaulted_history_rechecks_late_tool_credential() { + let vault = CredentialVault::new(); + let raw = "late-shared-tool-secret-1234567890"; + let history = Arc::new(vec![Message { + role: Role::User, + content: MessageContent::text(format!("Use {raw}")), + }]); + let first = vault_provider_history_shared(&history, &vault).unwrap(); + assert!(serde_json::to_string(first.as_ref()).unwrap().contains(raw)); + + let config = RequestConfig { + tools: Arc::new(vec![Tool { + name: "safe_tool".to_owned(), + description: format!("token: {raw}"), + input_schema: serde_json::json!({"type": "object"}), + schema_enforcement: Default::default(), + }]), + ..RequestConfig::default() + }; + let request = ProviderSafeRequest::prepare_vaulted(first, config, &vault).unwrap(); + let wire = serde_json::to_string(request.messages.as_ref()).unwrap(); + assert!(wire.contains("{{CRED|")); + assert!(!wire.contains(raw)); + request.ensure_current(&vault).unwrap(); +} + #[tokio::test] async fn test_wait_for_tool_response_buffers_out_of_order() { let (tx, rx) = mpsc::unbounded_channel(); diff --git a/packages/tui-rs/Cargo.toml b/packages/tui-rs/Cargo.toml index c5a94df7a..97da86837 100644 --- a/packages/tui-rs/Cargo.toml +++ b/packages/tui-rs/Cargo.toml @@ -175,7 +175,7 @@ windows-sys = { workspace = true, features = [ default = [] # Unverified NativeAgent constructor for in-crate tests plus the local # perf/bench binaries. Shipped maestro / deixic-code builds must not enable it. -test-support = ["maestro-local-host/test-support"] +test-support = ["maestro-local-host/test-support", "maestro-runtime/test-support"] thin-scenario = ["maestro-local-host/thin-scenario"] clipboard = ["arboard"] # Hook backends diff --git a/packages/tui-rs/benches/baselines/README.md b/packages/tui-rs/benches/baselines/README.md index dae4b3921..4de97c008 100644 --- a/packages/tui-rs/benches/baselines/README.md +++ b/packages/tui-rs/benches/baselines/README.md @@ -8,6 +8,27 @@ scenario that regresses by more than 15% (`--threshold` overrides). The gate is **advisory**: the `perf-baselines` workflow warns and fails open, and is not a required status check. +The separate `work-counts.json` records deterministic work for 96 scripted +text turns with a growing history. Each prompt is 365–366 ASCII bytes and each +response is 334–335 ASCII bytes. The fixture uses `ScriptedClient`, requires +96 completed turns and zero tool ends, and issues no tool calls or file-tool +dispatches. +`provider_history_vault_passes` counts full-history credential scans: 192 +with two passes per provider request. The benchmark emits the fixture and +metric as JSON with `--work-counts-json`, so CI can enforce a fixed work ceiling +without relying on machine-dependent timing. The 32-tool-turn timing scenario +remains advisory because file-tool dispatch obscures this request-preparation +cost. + +In four balanced candidate/control pairs on a busy 16-core Linux host (debug +build), the candidate's 96-turn wall-clock p50/p75/p95 was +32.49/34.18/35.73 seconds versus 36.00/37.09/39.57 seconds for the old +three-pass path. User CPU p50/p75/p95 was 22.15/23.19/25.04 seconds versus +26.01/26.35/26.70 seconds. All four paired wall and CPU results favored the +two-pass path. The p95 values are interpolated from only four samples; these +synthetic provider-loop timings do not measure user-visible latency. The old +path made 288 full-history vault passes on the same fixture, versus 192 now. + File naming: `.json` where `` is `-` — `linux-x86_64`, `linux-aarch64`, `macos-aarch64`. @@ -30,6 +51,7 @@ File naming: `.json` where `` is `-` — ``` cargo run -p maestro-tui --release --locked --features test-support --bin maestro-perf-bench +cargo run -p maestro-tui --locked --features test-support --bin maestro-perf-bench -- --work-counts-json ``` ## Producing or refreshing a baseline @@ -53,8 +75,8 @@ cargo run -p maestro-tui --release --locked --features test-support --bin maestr ``` Exits 1 and prints the regressed scenarios when any slowdown exceeds the -threshold; a missing baseline file fails loudly with instructions. Scenarios -present on only one side are skipped. +threshold; a missing baseline file or a required scenario on either side fails +loudly with instructions. ## Notes diff --git a/packages/tui-rs/benches/baselines/work-counts.json b/packages/tui-rs/benches/baselines/work-counts.json new file mode 100644 index 000000000..a9fea2db7 --- /dev/null +++ b/packages/tui-rs/benches/baselines/work-counts.json @@ -0,0 +1,6 @@ +{ + "fixture": "agent_loop_96_long_history_turns", + "metrics": { + "provider_history_vault_passes": 192 + } +} diff --git a/packages/tui-rs/src/bin/maestro_perf_bench.rs b/packages/tui-rs/src/bin/maestro_perf_bench.rs index aee20e5fb..f445520c0 100644 --- a/packages/tui-rs/src/bin/maestro_perf_bench.rs +++ b/packages/tui-rs/src/bin/maestro_perf_bench.rs @@ -21,6 +21,7 @@ use std::path::{Path, PathBuf}; use std::time::{Instant, SystemTime}; use anyhow::{Context, Result, bail}; +use maestro_runtime::agent::take_provider_history_vault_passes_for_bench; use maestro_tui::agent::{FromAgent, NativeAgent, NativeAgentConfig}; use maestro_tui::ai::{ScriptedBlock, ScriptedClient, ScriptedResponse, StopReason, UnifiedClient}; use maestro_tui::components::{ChatView, ModelSelector}; @@ -81,8 +82,8 @@ impl Comparison { } } -/// Compare current timings against a baseline. Scenarios absent from either -/// side are skipped; only shared scenarios can regress. +/// Compare current timings against a baseline. Call +/// `require_baseline_scenarios` first when enforcing a baseline. fn compare(baseline: &BTreeMap, current: &BTreeMap) -> Vec { baseline .iter() @@ -96,6 +97,20 @@ fn compare(baseline: &BTreeMap, current: &BTreeMap) -> .collect() } +fn require_baseline_scenarios( + baseline: &BTreeMap, + current: &BTreeMap, +) -> Result<()> { + let missing = baseline + .keys() + .filter(|name| !current.contains_key(*name)) + .collect::>(); + if !missing.is_empty() { + bail!("required timing baseline scenarios are missing: {missing:?}"); + } + Ok(()) +} + /// Scenarios whose slowdown exceeds `threshold` (e.g. 0.15 for 15%). fn regressions(comparisons: &[Comparison], threshold: f64) -> Vec<&Comparison> { comparisons @@ -658,6 +673,7 @@ struct Args { write_baseline: Option, baseline: Option, threshold: f64, + work_counts_json: bool, } fn parse_args() -> Result { @@ -665,6 +681,7 @@ fn parse_args() -> Result { write_baseline: None, baseline: None, threshold: DEFAULT_THRESHOLD, + work_counts_json: false, }; let mut iter = std::env::args().skip(1); while let Some(arg) = iter.next() { @@ -685,9 +702,10 @@ fn parse_args() -> Result { .parse() .with_context(|| format!("invalid --threshold value: {value}"))?; } + "--work-counts-json" => args.work_counts_json = true, "-h" | "--help" => { println!( - "maestro-perf-bench [--write-baseline ] [--baseline ] [--threshold ]" + "maestro-perf-bench [--write-baseline ] [--baseline ] [--threshold ] [--work-counts-json]" ); std::process::exit(0); } @@ -697,11 +715,29 @@ fn parse_args() -> Result { if args.write_baseline.is_some() && args.baseline.is_some() { bail!("--write-baseline and --baseline are mutually exclusive"); } + if args.work_counts_json && (args.write_baseline.is_some() || args.baseline.is_some()) { + bail!("--work-counts-json cannot be combined with timing baselines"); + } Ok(args) } fn main() -> Result<()> { let args = parse_args()?; + if args.work_counts_json { + take_provider_history_vault_passes_for_bench(); + Runtime::new() + .context("create work-count runtime")? + .block_on(run_scripted_long_history(AGENT_LONG_HISTORY_TURN_COUNT)); + let passes = take_provider_history_vault_passes_for_bench(); + println!( + "{}", + serde_json::json!({ + "fixture": "agent_loop_96_long_history_turns", + "metrics": {"provider_history_vault_passes": passes} + }) + ); + return Ok(()); + } let current = run_scenarios()?; if let Some(path) = args.write_baseline { @@ -718,6 +754,7 @@ fn main() -> Result<()> { }; let baseline = load_baseline(&path)?; + require_baseline_scenarios(&baseline.scenarios, ¤t)?; let comparisons = compare(&baseline.scenarios, ¤t); println!( @@ -805,6 +842,8 @@ mod tests { let comparisons = compare(&baseline, ¤t); assert_eq!(comparisons.len(), 1); assert_eq!(comparisons[0].name, "a"); + assert!(require_baseline_scenarios(&baseline, ¤t).is_err()); + assert!(require_baseline_scenarios(¤t, &baseline).is_err()); } #[test] diff --git a/packages/tui-rs/src/update_cli.rs b/packages/tui-rs/src/update_cli.rs index cb83744b3..ec227da2c 100644 --- a/packages/tui-rs/src/update_cli.rs +++ b/packages/tui-rs/src/update_cli.rs @@ -1172,6 +1172,7 @@ fn package_install_context_from( manager_override: Option<&str>, ) -> Option { let package_root = dunce::canonicalize(package_root).ok()?; + let executable = dunce::canonicalize(executable).ok()?; let relative = executable.strip_prefix(&package_root).ok()?; let components = relative .components() @@ -3731,6 +3732,16 @@ mod tests { .join("bin/maestro"), } ); + let aliased_executable = alias.join("vendor/maestro/test-target/maestro"); + assert_eq!( + package_install_context_from( + &aliased_executable, + &package_root, + "@evalops/maestro".to_owned(), + Some("npm"), + ), + Some(context), + ); } #[test]