diff --git a/Cargo.toml b/Cargo.toml index 0e0335b..a2fd488 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "animus-provider-oai-agent" -version = "0.1.5" +version = "0.1.6" edition = "2021" license = "Elastic-2.0" description = "Full agentic OpenAI-compatible provider plugin for Animus — tool calling, MCP, multi-turn agent loop, file ops, bash, search" diff --git a/plugin.toml b/plugin.toml index 9205394..4cb657e 100644 --- a/plugin.toml +++ b/plugin.toml @@ -1,6 +1,6 @@ schema = "animus.plugin.v1" name = "animus-provider-oai-agent" -version = "0.1.5" +version = "0.1.6" plugin_kind = "provider" description = "Full agentic OpenAI-compatible provider plugin for Animus — tool calling, MCP, multi-turn agent loop" diff --git a/src/bin/animus-oai-runner.rs b/src/bin/animus-oai-runner.rs index 5daef9d..0d67ca2 100644 --- a/src/bin/animus-oai-runner.rs +++ b/src/bin/animus-oai-runner.rs @@ -132,10 +132,19 @@ async fn main() -> Result<()> { None => None, }; - let client = api::client::ApiClient::new( + // Resolve provider-side caching from env (the runtime contract has no + // `cache` object plumbed yet — ao-cli will wire that seam later; the + // env fallbacks make it controllable today). + let cache_config = config::CacheConfig::resolve(None); + let response_cache_ttl = cache_config + .response_cache + .then_some(cache_config.response_cache_ttl_secs); + + let client = api::client::ApiClient::with_response_cache( resolved_config.api_base, resolved_config.api_key, idle_timeout, + response_cache_ttl, ); let native_tools = if read_only { @@ -206,6 +215,7 @@ async fn main() -> Result<()> { cancel_token, context_limit, max_tokens, + cache_config.prompt_breakpoints, ) .await; diff --git a/src/runner_core/api/client.rs b/src/runner_core/api/client.rs index 852dfb9..c9002a0 100644 --- a/src/runner_core/api/client.rs +++ b/src/runner_core/api/client.rs @@ -156,10 +156,25 @@ pub struct ApiClient { http: reqwest::Client, api_base: String, api_key: String, + /// When set, opt into OpenRouter's response cache via request headers. + /// `Some(ttl_secs)` sends `X-OpenRouter-Cache: true` plus + /// `X-OpenRouter-Cache-TTL: `; `None` sends neither. + response_cache_ttl_secs: Option, } impl ApiClient { pub fn new(api_base: String, api_key: String, timeout_secs: u64) -> Self { + Self::with_response_cache(api_base, api_key, timeout_secs, None) + } + + /// Construct a client, optionally enabling OpenRouter's response cache with + /// the given TTL (seconds). `None` leaves response caching off. + pub fn with_response_cache( + api_base: String, + api_key: String, + timeout_secs: u64, + response_cache_ttl_secs: Option, + ) -> Self { let http = reqwest::Client::builder() .timeout(Duration::from_secs(timeout_secs)) .build() @@ -168,9 +183,17 @@ impl ApiClient { http, api_base, api_key, + response_cache_ttl_secs, } } + /// The configured chat-completions base URL (e.g. the OpenRouter or OpenAI + /// endpoint). Used to decide whether Anthropic/OpenRouter prompt-cache + /// breakpoints apply. + pub fn api_base(&self) -> &str { + &self.api_base + } + pub async fn stream_chat( &self, request: &ChatRequest, @@ -269,15 +292,31 @@ impl ApiClient { request: &ChatRequest, on_text_chunk: &mut dyn FnMut(&str), ) -> Result<(ChatMessage, Option)> { - let resp = self + let mut builder = self .http .post(url) .header("Authorization", format!("Bearer {}", self.api_key)) .header("Content-Type", "application/json") - .header("User-Agent", "claude-code/2.1.80") - .json(request) - .send() - .await?; + .header("User-Agent", "claude-code/2.1.80"); + // Opt into OpenRouter's response cache (byte-identical request reuse, + // billed as 0 tokens on a hit) only when explicitly enabled. + if let Some(ttl) = self.response_cache_ttl_secs { + builder = builder + .header("X-OpenRouter-Cache", "true") + .header("X-OpenRouter-Cache-TTL", ttl.to_string()); + } + let resp = builder.json(request).send().await?; + + // Surface OpenRouter's response-cache hit/miss at debug for observability. + if self.response_cache_ttl_secs.is_some() { + if let Some(cache_status) = resp + .headers() + .get("x-openrouter-cache-status") + .and_then(|v| v.to_str().ok()) + { + eprintln!("[oai-runner] OpenRouter response cache: {}", cache_status); + } + } let status = resp.status(); if !status.is_success() { @@ -457,6 +496,72 @@ mod tests { } } + // ── response-cache headers (P2) ───────────────────────────────── + + fn minimal_request() -> ChatRequest { + ChatRequest { + model: "anthropic/claude-sonnet-4".to_string(), + messages: vec![ChatMessage { + reasoning_content: None, + role: "user".to_string(), + content: Some("hi".to_string()), + tool_calls: None, + tool_call_id: None, + }], + stream: true, + tools: None, + max_tokens: Some(16), + response_format: None, + stream_options: None, + cache_message_breakpoints: Vec::new(), + } + } + + /// A single SSE chunk + [DONE] so `do_stream` returns cleanly. + const SSE_BODY: &str = + "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\ndata: [DONE]\n\n"; + + #[tokio::test] + async fn response_cache_headers_sent_when_enabled() { + let mut server = mockito::Server::new_async().await; + let mock = server + .mock("POST", "/chat/completions") + .match_header("x-openrouter-cache", "true") + .match_header("x-openrouter-cache-ttl", "600") + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_header("x-openrouter-cache-status", "MISS") + .with_body(SSE_BODY) + .create_async() + .await; + + let client = + ApiClient::with_response_cache(server.url(), "sk-test".to_string(), 5, Some(600)); + let res = client.stream_chat(&minimal_request(), &mut |_| {}).await; + assert!(res.is_ok(), "stream should succeed: {res:?}"); + mock.assert_async().await; // header matchers verified + } + + #[tokio::test] + async fn response_cache_headers_absent_when_disabled() { + let mut server = mockito::Server::new_async().await; + // Expect NO cache headers on the request. + let mock = server + .mock("POST", "/chat/completions") + .match_header("x-openrouter-cache", mockito::Matcher::Missing) + .match_header("x-openrouter-cache-ttl", mockito::Matcher::Missing) + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_body(SSE_BODY) + .create_async() + .await; + + let client = ApiClient::new(server.url(), "sk-test".to_string(), 5); + let res = client.stream_chat(&minimal_request(), &mut |_| {}).await; + assert!(res.is_ok(), "stream should succeed: {res:?}"); + mock.assert_async().await; + } + // ── circuit_state ────────────────────────────────────────────── #[test] diff --git a/src/runner_core/api/types.rs b/src/runner_core/api/types.rs index 4b1f396..d5c18ba 100644 --- a/src/runner_core/api/types.rs +++ b/src/runner_core/api/types.rs @@ -1,6 +1,78 @@ -use serde::{Deserialize, Serialize}; +use serde::ser::{SerializeMap, SerializeSeq}; +use serde::{Deserialize, Serialize, Serializer}; use serde_json::Value; +/// Anthropic-via-OpenRouter prompt-cache breakpoint marker. +/// +/// Serializes to `{"type":"ephemeral"}` and is attached to a content part to +/// tell OpenRouter/Anthropic where the cacheable prefix ends. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct CacheControl { + #[serde(rename = "type")] + pub type_: String, +} + +impl CacheControl { + pub fn ephemeral() -> Self { + Self { + type_: "ephemeral".to_string(), + } + } +} + +/// Decide whether a resolved model + endpoint needs an explicit +/// `cache_control` breakpoint to engage prompt caching. +/// +/// Some model families served through OpenRouter cache ONLY when a content part +/// carries `cache_control` (caching is not automatic). Per OpenRouter's +/// prompt-caching docs, exactly TWO families need explicit breakpoints: +/// **Anthropic Claude** and **Alibaba Qwen**. OpenAI, DeepSeek, Grok/xAI, +/// Moonshot/Kimi, Groq, and Gemini 2.5 all cache automatically and must stay +/// plain strings (an unexpected `cache_control` field can be rejected). +/// +/// We require BOTH an explicit-cache model AND an OpenRouter route before +/// attaching the breakpoint: +/// +/// - Plain OpenAI-native (and other auto-cache) endpoints must stay plain. +/// - A custom non-OpenRouter, OpenAI-compatible endpoint that happens to be fed +/// an `anthropic/...` or `qwen/...` model id must NOT receive the +/// OpenRouter-specific multipart shape either — hence the explicit OpenRouter +/// gate rather than treating the model string alone as sufficient. +pub fn wants_explicit_cache_breakpoint(model: &str, api_base: &str) -> bool { + let m = model.to_ascii_lowercase(); + let base = api_base.to_ascii_lowercase(); + // OpenRouter namespaces a model as `/` (optionally behind its own + // `openrouter/` prefix). Match on the vendor SEGMENT rather than a bare + // substring so we don't trip on auto-cache models that merely mention an + // explicit-cache family in their id (e.g. the DeepSeek-derived + // `deepseek/deepseek-r1-distill-qwen-32b`, which must stay plain). + let vendor = m + .strip_prefix("openrouter/") + .unwrap_or(&m) + .split('/') + .next() + .unwrap_or(""); + // Families that require an explicit `cache_control` breakpoint on OpenRouter. + // Anthropic and Alibaba Qwen are the only two; everything else auto-caches. + let needs_explicit_cache = vendor == "anthropic" + || vendor == "qwen" + || vendor == "alibaba" + // Bare model names with no vendor segment (e.g. "claude-3-5-sonnet"). + || (vendor == m && m.contains("claude")); + // The request is going through the OpenRouter gateway when the base URL + // points at openrouter, or the model id carries OpenRouter's own + // `openrouter/` namespace prefix. + let via_openrouter = base.contains("openrouter") || m.starts_with("openrouter/"); + needs_explicit_cache && via_openrouter +} + +/// Deprecated alias kept for API compatibility. The gate was broadened beyond +/// Anthropic to also cover Alibaba Qwen; use [`wants_explicit_cache_breakpoint`]. +#[deprecated(note = "renamed to `wants_explicit_cache_breakpoint` (now also covers Qwen)")] +pub fn wants_anthropic_cache_breakpoint(model: &str, api_base: &str) -> bool { + wants_explicit_cache_breakpoint(model, api_base) +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ChatMessage { pub role: String, @@ -120,19 +192,138 @@ pub struct JsonSchemaSpec { pub schema: Value, } -#[derive(Debug, Clone, Serialize)] +/// Anthropic caps explicit `cache_control` breakpoints at 4 across the whole +/// prompt (tools, system, and messages combined). We never emit more than this. +pub const MAX_CACHE_BREAKPOINTS: usize = 4; + +#[derive(Debug, Clone)] pub struct ChatRequest { pub model: String, pub messages: Vec, pub stream: bool, - #[serde(skip_serializing_if = "Option::is_none")] pub tools: Option>, - #[serde(skip_serializing_if = "Option::is_none")] pub max_tokens: Option, - #[serde(skip_serializing_if = "Option::is_none")] pub response_format: Option, - #[serde(skip_serializing_if = "Option::is_none")] pub stream_options: Option, + /// Indices of messages that should carry an Anthropic/OpenRouter + /// `cache_control:{type:"ephemeral"}` breakpoint. The marked messages + /// serialize their `content` as the multipart array form; everything else + /// stays a plain string. Empty = no prompt caching (auto-cache families and + /// non-OpenRouter routes). Tools are cached implicitly: Anthropic orders the + /// prompt as tools -> system -> messages, so a breakpoint on the `system` + /// message already caches the tools prefix. Capped at + /// [`MAX_CACHE_BREAKPOINTS`]; populate via + /// `agent_loop::plan_cache_breakpoints`. Not serialized as a field of its + /// own — it only changes how the marked messages' `content` is emitted. + pub cache_message_breakpoints: Vec, +} + +/// A single text content part carrying an optional cache breakpoint, used to +/// emit the multipart `content` array form Anthropic-via-OpenRouter requires. +#[derive(Serialize)] +struct CachedTextPart<'a> { + #[serde(rename = "type")] + type_: &'static str, + text: &'a str, + cache_control: CacheControl, +} + +/// Serializes a single message with a cache breakpoint: its `content` becomes +/// the multipart array form while every OTHER field (role, reasoning_content, +/// tool_calls, tool_call_id) is preserved exactly as the derived `ChatMessage` +/// path would emit it. The caller guarantees `content` is `Some` and non-empty. +struct CachedMessage<'a> { + msg: &'a ChatMessage, +} + +impl Serialize for CachedMessage<'_> { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + let mut map = serializer.serialize_map(None)?; + map.serialize_entry("role", &self.msg.role)?; + if let Some(text) = self.msg.content.as_deref() { + map.serialize_entry( + "content", + &[CachedTextPart { + type_: "text", + text, + cache_control: CacheControl::ephemeral(), + }], + )?; + } + if let Some(reasoning) = &self.msg.reasoning_content { + map.serialize_entry("reasoning_content", reasoning)?; + } + if let Some(tool_calls) = &self.msg.tool_calls { + map.serialize_entry("tool_calls", tool_calls)?; + } + if let Some(tool_call_id) = &self.msg.tool_call_id { + map.serialize_entry("tool_call_id", tool_call_id)?; + } + map.end() + } +} + +/// Serializes the message sequence, rewriting the messages at +/// `cache_breakpoints` into the multipart cache-breakpoint form. All other +/// messages serialize via the plain derived `ChatMessage` path (stable +/// plain-string `content`), so only the cacheable prefixes change shape. A +/// breakpoint index is honored only when that message has non-empty `content`. +struct MessagesView<'a> { + messages: &'a [ChatMessage], + cache_breakpoints: &'a [usize], +} + +impl Serialize for MessagesView<'_> { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + let mut seq = serializer.serialize_seq(Some(self.messages.len()))?; + for (i, msg) in self.messages.iter().enumerate() { + let cacheable = self.cache_breakpoints.contains(&i) + && msg.content.as_deref().is_some_and(|c| !c.is_empty()); + if cacheable { + seq.serialize_element(&CachedMessage { msg })?; + } else { + seq.serialize_element(msg)?; + } + } + seq.end() + } +} + +impl Serialize for ChatRequest { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + let mut map = serializer.serialize_map(None)?; + map.serialize_entry("model", &self.model)?; + map.serialize_entry( + "messages", + &MessagesView { + messages: &self.messages, + cache_breakpoints: &self.cache_message_breakpoints, + }, + )?; + map.serialize_entry("stream", &self.stream)?; + if let Some(tools) = &self.tools { + map.serialize_entry("tools", tools)?; + } + if let Some(max_tokens) = &self.max_tokens { + map.serialize_entry("max_tokens", max_tokens)?; + } + if let Some(response_format) = &self.response_format { + map.serialize_entry("response_format", response_format)?; + } + if let Some(stream_options) = &self.stream_options { + map.serialize_entry("stream_options", stream_options)?; + } + map.end() + } } #[derive(Debug, Clone, Serialize)] @@ -161,6 +352,7 @@ mod tests { max_tokens: Some(4096), response_format: None, stream_options: None, + cache_message_breakpoints: Vec::new(), }; let json = serde_json::to_value(&request).unwrap(); assert_eq!(json["model"], "minimax/MiniMax-M2.1"); @@ -192,6 +384,7 @@ mod tests { }), }), stream_options: None, + cache_message_breakpoints: Vec::new(), }; let json = serde_json::to_value(&request).unwrap(); assert_eq!(json["response_format"]["type"], "json_schema"); @@ -343,4 +536,271 @@ mod tests { let usage: UsageInfo = serde_json::from_str(raw).unwrap(); assert_eq!(usage.cache_read_input_tokens, 0); } + + /// Build a `[system, user]` request. When `cache_system` is true, mark only + /// the system message (index 0) as a breakpoint — exercising the + /// single-breakpoint serialization path. + fn sys_then_user_request(model: &str, cache_system: bool) -> ChatRequest { + ChatRequest { + model: model.to_string(), + messages: vec![ + ChatMessage { + reasoning_content: None, + role: "system".to_string(), + content: Some("You are a helpful agent.".to_string()), + tool_calls: None, + tool_call_id: None, + }, + ChatMessage { + reasoning_content: None, + role: "user".to_string(), + content: Some("do the thing".to_string()), + tool_calls: None, + tool_call_id: None, + }, + ], + stream: true, + tools: None, + max_tokens: Some(4096), + response_format: None, + stream_options: None, + cache_message_breakpoints: if cache_system { vec![0] } else { Vec::new() }, + } + } + + #[test] + fn gate_true_for_explicit_cache_families_via_openrouter() { + // Anthropic Claude — explicit-cache family. + assert!(wants_explicit_cache_breakpoint( + "anthropic/claude-sonnet-4", + "https://openrouter.ai/api/v1" + )); + assert!(wants_explicit_cache_breakpoint( + "openrouter/anthropic/claude-3.5-sonnet", + "https://openrouter.ai/api/v1" + )); + // `claude` model name routed through an openrouter base_url. + assert!(wants_explicit_cache_breakpoint( + "claude-3-5-sonnet", + "https://openrouter.ai/api/v1" + )); + // Alibaba Qwen — the second explicit-cache family. + assert!(wants_explicit_cache_breakpoint( + "qwen/qwen-2.5-72b-instruct", + "https://openrouter.ai/api/v1" + )); + assert!(wants_explicit_cache_breakpoint( + "openrouter/qwen/qwen-2.5-72b-instruct", + "https://openrouter.ai/api/v1" + )); + } + + #[test] + fn gate_false_for_auto_cache_families_and_off_openrouter() { + // Plain OpenAI-native endpoint must NOT get cache_control. + assert!(!wants_explicit_cache_breakpoint( + "gpt-4o", + "https://api.openai.com/v1" + )); + // Auto-cache families on OpenRouter must stay plain (no breakpoint): + // DeepSeek, Gemini 2.5, Grok/xAI, Moonshot/Kimi, Groq, plain OpenAI. + for model in [ + "deepseek/deepseek-chat", + // DeepSeek-derived model whose id merely mentions qwen: stays auto-cache. + "deepseek/deepseek-r1-distill-qwen-32b", + "google/gemini-2.5-pro", + "x-ai/grok-2", + "moonshotai/kimi-k2", + "groq/llama-3.3-70b", + "openai/gpt-4o", + "minimax/MiniMax-M2.1", + ] { + assert!( + !wants_explicit_cache_breakpoint(model, "https://openrouter.ai/api/v1"), + "auto-cache model {model} must NOT get a cache_control breakpoint" + ); + } + // A `claude`-named model on a non-openrouter base should not trip. + assert!(!wants_explicit_cache_breakpoint( + "claude-3-5-sonnet", + "https://api.openai.com/v1" + )); + // Explicit-cache model ids pointed at a custom non-openrouter, + // OpenAI-compatible base must NOT get the OpenRouter-specific shape. + assert!(!wants_explicit_cache_breakpoint( + "anthropic/claude-sonnet-4", + "https://my-proxy.example.com/v1" + )); + assert!(!wants_explicit_cache_breakpoint( + "qwen/qwen-2.5-72b-instruct", + "https://my-proxy.example.com/v1" + )); + } + + #[test] + fn anthropic_request_serializes_system_as_multipart_with_cache_control() { + let request = sys_then_user_request("anthropic/claude-sonnet-4", true); + let json = serde_json::to_value(&request).unwrap(); + + let sys = &json["messages"][0]; + assert_eq!(sys["role"], "system"); + // content is the multipart array form, not a plain string. + let parts = sys["content"].as_array().expect("system content is array"); + assert_eq!(parts.len(), 1); + assert_eq!(parts[0]["type"], "text"); + assert_eq!(parts[0]["text"], "You are a helpful agent."); + assert_eq!(parts[0]["cache_control"]["type"], "ephemeral"); + + // The downstream user message is untouched (plain string content). + let user = &json["messages"][1]; + assert_eq!(user["role"], "user"); + assert_eq!(user["content"], "do the thing"); + assert!(user["content"].is_string()); + assert!(user.get("cache_control").is_none()); + } + + #[test] + fn qwen_request_serializes_system_as_multipart_with_cache_control() { + // Drive the flag from the real gate to prove Qwen-via-OpenRouter routes + // through the multipart breakpoint path end to end. + let model = "qwen/qwen-2.5-72b-instruct"; + let cache = wants_explicit_cache_breakpoint(model, "https://openrouter.ai/api/v1"); + assert!(cache, "Qwen-via-OpenRouter should request a breakpoint"); + let request = sys_then_user_request(model, cache); + let json = serde_json::to_value(&request).unwrap(); + + let parts = json["messages"][0]["content"] + .as_array() + .expect("system content is array"); + assert_eq!(parts.len(), 1); + assert_eq!(parts[0]["text"], "You are a helpful agent."); + assert_eq!(parts[0]["cache_control"]["type"], "ephemeral"); + } + + #[test] + fn openai_native_request_keeps_system_as_plain_string() { + let request = sys_then_user_request("gpt-4o", false); + let json = serde_json::to_value(&request).unwrap(); + + let sys = &json["messages"][0]; + assert_eq!(sys["role"], "system"); + // Stays a plain string; no cache_control anywhere. + assert!(sys["content"].is_string()); + assert_eq!(sys["content"], "You are a helpful agent."); + assert!(sys.get("cache_control").is_none()); + assert!(json["messages"][1].get("cache_control").is_none()); + } + + #[test] + fn only_first_system_message_gets_breakpoint() { + let mut request = sys_then_user_request("anthropic/claude-sonnet-4", true); + // Insert a second system message; only the first should be rewritten. + request.messages.insert( + 1, + ChatMessage { + reasoning_content: None, + role: "system".to_string(), + content: Some("second system".to_string()), + tool_calls: None, + tool_call_id: None, + }, + ); + let json = serde_json::to_value(&request).unwrap(); + assert!(json["messages"][0]["content"].is_array()); + assert!(json["messages"][1]["content"].is_string()); + assert_eq!(json["messages"][1]["content"], "second system"); + } + + #[test] + fn cached_request_body_round_trips_to_expected_shape() { + let request = sys_then_user_request("anthropic/claude-sonnet-4", true); + let body = serde_json::to_string(&request).unwrap(); + let reparsed: Value = serde_json::from_str(&body).unwrap(); + assert_eq!( + reparsed["messages"][0]["content"][0]["cache_control"], + json!({ "type": "ephemeral" }) + ); + assert_eq!(reparsed["model"], "anthropic/claude-sonnet-4"); + assert_eq!(reparsed["stream"], true); + } + + #[test] + fn multiple_breakpoints_mark_exactly_the_requested_indices() { + // system(0), user(1), assistant(2), user(3): mark system + the two + // user turns as a multi-breakpoint plan; assistant stays plain. + let mut request = sys_then_user_request("anthropic/claude-sonnet-4", false); + request.messages.push(ChatMessage { + reasoning_content: None, + role: "assistant".to_string(), + content: Some("interim answer".to_string()), + tool_calls: None, + tool_call_id: None, + }); + request.messages.push(ChatMessage { + reasoning_content: None, + role: "user".to_string(), + content: Some("follow up".to_string()), + tool_calls: None, + tool_call_id: None, + }); + request.cache_message_breakpoints = vec![0, 1, 3]; + let json = serde_json::to_value(&request).unwrap(); + + // Marked indices are multipart with an ephemeral breakpoint. + for i in [0usize, 1, 3] { + let parts = json["messages"][i]["content"] + .as_array() + .unwrap_or_else(|| panic!("message {i} should be multipart")); + assert_eq!(parts[0]["cache_control"]["type"], "ephemeral"); + } + // Unmarked assistant message stays a plain string. + assert!(json["messages"][2]["content"].is_string()); + } + + #[test] + fn breakpoint_on_message_without_content_is_skipped() { + // An assistant tool-call message with no text content can't carry a + // text part; marking it must NOT produce an empty/invalid content array. + let mut request = sys_then_user_request("anthropic/claude-sonnet-4", false); + request.messages.push(ChatMessage { + reasoning_content: None, + role: "assistant".to_string(), + content: None, + tool_calls: Some(vec![ToolCall { + id: "call_1".to_string(), + type_: "function".to_string(), + function: FunctionCall { + name: "read_file".to_string(), + arguments: "{}".to_string(), + }, + }]), + tool_call_id: None, + }); + request.cache_message_breakpoints = vec![2]; + let json = serde_json::to_value(&request).unwrap(); + // No content key emitted (skip_serializing_if None) and no cache_control. + assert!(json["messages"][2].get("content").is_none()); + assert!(json["messages"][2]["tool_calls"].is_array()); + } + + #[test] + fn tool_role_breakpoint_preserves_tool_call_id() { + // A `tool` result message marked cacheable keeps its tool_call_id while + // its content becomes the multipart breakpoint form. + let mut request = sys_then_user_request("anthropic/claude-sonnet-4", false); + request.messages.push(ChatMessage { + reasoning_content: None, + role: "tool".to_string(), + content: Some("file contents".to_string()), + tool_calls: None, + tool_call_id: Some("call_1".to_string()), + }); + request.cache_message_breakpoints = vec![2]; + let json = serde_json::to_value(&request).unwrap(); + let msg = &json["messages"][2]; + assert_eq!(msg["role"], "tool"); + assert_eq!(msg["tool_call_id"], "call_1"); + assert_eq!(msg["content"][0]["cache_control"]["type"], "ephemeral"); + assert_eq!(msg["content"][0]["text"], "file contents"); + } } diff --git a/src/runner_core/config.rs b/src/runner_core/config.rs index c0609c8..e70445b 100644 --- a/src/runner_core/config.rs +++ b/src/runner_core/config.rs @@ -1,5 +1,6 @@ use anyhow::{bail, Result}; use serde::Deserialize; +use serde_json::Value; use std::collections::HashMap; use std::path::PathBuf; @@ -9,6 +10,88 @@ pub enum StructuredOutputSupport { JsonObjectOnly, } +/// OpenRouter default response-cache TTL when none is supplied (seconds). +pub const DEFAULT_RESPONSE_CACHE_TTL_SECS: u32 = 300; +/// OpenRouter response-cache TTL bounds (seconds). +pub const RESPONSE_CACHE_TTL_MIN_SECS: u32 = 1; +pub const RESPONSE_CACHE_TTL_MAX_SECS: u32 = 86_400; + +/// Provider-side caching configuration, read from the per-request runtime +/// contract's optional `cache` object and overridable via env vars. +/// +/// Defaults: prompt-breakpoint caching is ON (pure savings — explicit +/// `cache_control` breakpoints only ever reduce cost), while OpenRouter +/// response caching is OFF (a re-run returning a byte-identical cached response +/// can surprise callers, so it must be opted into). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)] +#[serde(default, deny_unknown_fields)] +pub struct CacheConfig { + /// Place Anthropic/Qwen-via-OpenRouter `cache_control` breakpoints on the + /// stable prompt layers (system + rolling conversation). + pub prompt_breakpoints: bool, + /// Send the `X-OpenRouter-Cache` header to opt into OpenRouter's response + /// cache (byte-identical request reuse, billed as 0 tokens on a hit). + pub response_cache: bool, + /// TTL (seconds) for the OpenRouter response cache, clamped to + /// `[RESPONSE_CACHE_TTL_MIN_SECS, RESPONSE_CACHE_TTL_MAX_SECS]`. + pub response_cache_ttl_secs: u32, +} + +impl Default for CacheConfig { + fn default() -> Self { + Self { + prompt_breakpoints: true, + response_cache: false, + response_cache_ttl_secs: DEFAULT_RESPONSE_CACHE_TTL_SECS, + } + } +} + +impl CacheConfig { + /// Resolve the effective cache config from an optional runtime-contract + /// `cache` object, then apply env-var overrides. Env vars always win so + /// caching is controllable without an ao-cli change: + /// + /// * `OAI_AGENT_PROMPT_CACHE` — `1/true/on` / `0/false/off` + /// * `OAI_AGENT_RESPONSE_CACHE` — `1/true/on` / `0/false/off` + /// * `OAI_AGENT_RESPONSE_CACHE_TTL_SECS` — integer seconds + /// + /// When the contract omits `cache` and no env vars are set, the defaults + /// apply. The TTL is always clamped to the OpenRouter-supported range. + pub fn resolve(contract_cache: Option<&Value>) -> Self { + let mut cfg = contract_cache + .and_then(|v| serde_json::from_value::(v.clone()).ok()) + .unwrap_or_default(); + + if let Some(v) = env_bool("OAI_AGENT_PROMPT_CACHE") { + cfg.prompt_breakpoints = v; + } + if let Some(v) = env_bool("OAI_AGENT_RESPONSE_CACHE") { + cfg.response_cache = v; + } + if let Ok(raw) = std::env::var("OAI_AGENT_RESPONSE_CACHE_TTL_SECS") { + if let Ok(secs) = raw.trim().parse::() { + cfg.response_cache_ttl_secs = secs; + } + } + + cfg.response_cache_ttl_secs = cfg + .response_cache_ttl_secs + .clamp(RESPONSE_CACHE_TTL_MIN_SECS, RESPONSE_CACHE_TTL_MAX_SECS); + cfg + } +} + +/// Parse a boolean-ish env var; returns `None` when unset or unrecognized. +fn env_bool(name: &str) -> Option { + let raw = std::env::var(name).ok()?; + match raw.trim().to_ascii_lowercase().as_str() { + "1" | "true" | "on" | "yes" => Some(true), + "0" | "false" | "off" | "no" => Some(false), + _ => None, + } +} + pub struct ResolvedConfig { pub api_base: String, pub api_key: String, @@ -267,4 +350,89 @@ mod tests { assert_eq!(config.api_key, "sk-test-key"); assert_eq!(config.model_id, "MiniMax-M2.1"); } + + // `CacheConfig::resolve` reads process-global env vars. ALL resolve-based + // tests serialize on this lock and start from a cleared env so a test that + // sets env vars can't leak into a parallel reader. + static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + /// Lock the env, clear the three cache env vars, and return the guard. + fn lock_clean_env() -> std::sync::MutexGuard<'static, ()> { + let guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + std::env::remove_var("OAI_AGENT_PROMPT_CACHE"); + std::env::remove_var("OAI_AGENT_RESPONSE_CACHE"); + std::env::remove_var("OAI_AGENT_RESPONSE_CACHE_TTL_SECS"); + guard + } + + #[test] + fn cache_config_defaults_are_prompt_on_response_off() { + let cfg = CacheConfig::default(); + assert!(cfg.prompt_breakpoints, "prompt breakpoints default ON"); + assert!(!cfg.response_cache, "response cache default OFF"); + assert_eq!(cfg.response_cache_ttl_secs, DEFAULT_RESPONSE_CACHE_TTL_SECS); + } + + #[test] + fn cache_config_parses_partial_contract_object() { + let _guard = lock_clean_env(); + // Missing keys fall back to defaults (serde(default)). + let contract = serde_json::json!({ "response_cache": true }); + let cfg = CacheConfig::resolve(Some(&contract)); + assert!(cfg.response_cache); + assert!(cfg.prompt_breakpoints); // unspecified -> default on + assert_eq!(cfg.response_cache_ttl_secs, DEFAULT_RESPONSE_CACHE_TTL_SECS); + } + + #[test] + fn cache_config_clamps_ttl_to_supported_range() { + let _guard = lock_clean_env(); + let too_big = serde_json::json!({ "response_cache_ttl_secs": 999_999_u32 }); + assert_eq!( + CacheConfig::resolve(Some(&too_big)).response_cache_ttl_secs, + RESPONSE_CACHE_TTL_MAX_SECS + ); + let too_small = serde_json::json!({ "response_cache_ttl_secs": 0_u32 }); + assert_eq!( + CacheConfig::resolve(Some(&too_small)).response_cache_ttl_secs, + RESPONSE_CACHE_TTL_MIN_SECS + ); + } + + #[test] + fn cache_config_rejects_unknown_contract_fields_falling_back_to_default() { + let _guard = lock_clean_env(); + // deny_unknown_fields => parse fails => defaults used (no panic). + let contract = serde_json::json!({ "bogus": 1 }); + let cfg = CacheConfig::resolve(Some(&contract)); + assert_eq!(cfg, CacheConfig::default()); + } + + #[test] + fn cache_config_env_overrides_win_over_contract() { + let _guard = lock_clean_env(); + std::env::set_var("OAI_AGENT_PROMPT_CACHE", "off"); + std::env::set_var("OAI_AGENT_RESPONSE_CACHE", "true"); + std::env::set_var("OAI_AGENT_RESPONSE_CACHE_TTL_SECS", "600"); + // Contract says the opposite; env must win. + let contract = serde_json::json!({ + "prompt_breakpoints": true, + "response_cache": false, + "response_cache_ttl_secs": 300 + }); + let cfg = CacheConfig::resolve(Some(&contract)); + std::env::remove_var("OAI_AGENT_PROMPT_CACHE"); + std::env::remove_var("OAI_AGENT_RESPONSE_CACHE"); + std::env::remove_var("OAI_AGENT_RESPONSE_CACHE_TTL_SECS"); + + assert!(!cfg.prompt_breakpoints); + assert!(cfg.response_cache); + assert_eq!(cfg.response_cache_ttl_secs, 600); + } + + #[test] + fn cache_config_resolve_none_with_clean_env_is_default() { + let _guard = lock_clean_env(); + assert_eq!(CacheConfig::resolve(None), CacheConfig::default()); + } } diff --git a/src/runner_core/runner/agent_loop.rs b/src/runner_core/runner/agent_loop.rs index 6a500d6..131f13b 100644 --- a/src/runner_core/runner/agent_loop.rs +++ b/src/runner_core/runner/agent_loop.rs @@ -169,6 +169,81 @@ fn synthesize_fallback(model: &str, summary: &str, confidence: f64) -> Value { }) } +/// Plan which message indices should carry an Anthropic/OpenRouter +/// `cache_control` breakpoint for `messages`. +/// +/// When `enabled` is false (auto-cache family, non-OpenRouter route, or prompt +/// caching disabled by config) this returns an empty plan and the request +/// serializes every message as a plain string. +/// +/// Anthropic allows up to [`MAX_CACHE_BREAKPOINTS`] (4) breakpoints across the +/// whole prompt and caches the prefix tools -> system -> messages up to each +/// one. We follow Anthropic's recommended layout of static + rolling points: +/// +/// 1. The first non-empty `system` message — caches the stable tools+system +/// prefix (tools precede system, so they ride along for free). +/// 2. A ROLLING breakpoint on the most recent cacheable message — lets the next +/// step reuse the entire conversation so far from cache. +/// 3. A second rolling breakpoint ~one turn back (the last cacheable message +/// before the most recent `user` turn) for the sliding window, so the cache +/// survives an extra turn as it advances. +/// +/// Only messages with non-empty `content` and no `tool_calls` are eligible to +/// carry a breakpoint (those can be re-shaped into a single text content part +/// without dropping a tool-call payload). The result is sorted, de-duplicated, +/// and capped at 4. +pub fn plan_cache_breakpoints(messages: &[ChatMessage], enabled: bool) -> Vec { + if !enabled || messages.is_empty() { + return Vec::new(); + } + + let is_cacheable = |m: &ChatMessage| { + m.content.as_deref().is_some_and(|c| !c.is_empty()) && m.tool_calls.is_none() + }; + + let mut breakpoints: Vec = Vec::with_capacity(MAX_CACHE_BREAKPOINTS); + + // 1. Static: the first non-empty system message (caches tools + system). + let system_idx = messages + .iter() + .position(|m| m.role == "system" && m.content.as_deref().is_some_and(|c| !c.is_empty())); + if let Some(idx) = system_idx { + breakpoints.push(idx); + } + + // 2. Rolling: the most recent cacheable message overall. + if let Some(last) = (0..messages.len()) + .rev() + .find(|&i| Some(i) != system_idx && is_cacheable(&messages[i])) + { + if !breakpoints.contains(&last) { + breakpoints.push(last); + } + + // 3. Rolling: the last cacheable message strictly before the most recent + // `user` turn, for the sliding window. Skip if it collapses onto an + // already-chosen index. + let last_user = (0..messages.len()) + .rev() + .find(|&i| messages[i].role == "user"); + if let Some(user_idx) = last_user { + if let Some(prev_turn) = (0..user_idx) + .rev() + .find(|&i| Some(i) != system_idx && is_cacheable(&messages[i])) + { + if !breakpoints.contains(&prev_turn) { + breakpoints.push(prev_turn); + } + } + } + } + + breakpoints.sort_unstable(); + breakpoints.dedup(); + breakpoints.truncate(MAX_CACHE_BREAKPOINTS); + breakpoints +} + #[allow(clippy::too_many_arguments)] pub async fn run_agent_loop( client: &ApiClient, @@ -186,7 +261,13 @@ pub async fn run_agent_loop( cancel_token: CancellationToken, context_limit: usize, max_tokens: usize, + prompt_cache_enabled: bool, ) -> Result<()> { + // Anthropic/Qwen via OpenRouter need explicit breakpoints; auto-cache + // families and non-OpenRouter routes stay plain. Gated additionally by the + // provider's cache config (`prompt_cache_enabled`). + let cache_enabled = + prompt_cache_enabled && wants_explicit_cache_breakpoint(model, client.api_base()); let mut messages: Vec = Vec::new(); if let Some(sid) = session_id { @@ -326,6 +407,8 @@ pub async fn run_agent_loop( eprintln!( "[oai-runner] Context at capacity, compacting conversation history via LLM" ); + let compaction_breakpoints = + plan_cache_breakpoints(&compaction_msgs, cache_enabled); let compaction_request = ChatRequest { model: model.to_string(), messages: compaction_msgs, @@ -334,6 +417,7 @@ pub async fn run_agent_loop( max_tokens: Some(2048), response_format: None, stream_options: None, + cache_message_breakpoints: compaction_breakpoints, }; match client.stream_chat(&compaction_request, &mut |_| {}).await { Ok((summary_msg, _)) => { @@ -382,6 +466,7 @@ pub async fn run_agent_loop( stream_options: Some(StreamOptions { include_usage: true, }), + cache_message_breakpoints: plan_cache_breakpoints(&messages, cache_enabled), }; let (assistant_msg, usage) = client @@ -454,6 +539,7 @@ pub async fn run_agent_loop( &errors, output, structured_output, + cache_enabled, ) .await; if !corrected { @@ -728,6 +814,7 @@ async fn retry_schema_validation( initial_errors: &str, output: &mut OutputFormatter, structured_output: Option, + cache_enabled: bool, ) -> bool { let mut last_errors = initial_errors.to_string(); @@ -771,6 +858,10 @@ async fn retry_schema_validation( tool_call_id: None, }); + let retry_breakpoints = plan_cache_breakpoints( + &retry_messages, + cache_enabled && wants_explicit_cache_breakpoint(model, client.api_base()), + ); let retry_request = ChatRequest { model: model.to_string(), messages: retry_messages, @@ -784,6 +875,7 @@ async fn retry_schema_validation( stream_options: Some(StreamOptions { include_usage: true, }), + cache_message_breakpoints: retry_breakpoints, }; let retry_result = client @@ -893,6 +985,86 @@ mod tests { use super::*; use serde_json::json; + fn cm(role: &str, content: Option<&str>) -> ChatMessage { + ChatMessage { + reasoning_content: None, + role: role.to_string(), + content: content.map(|c| c.to_string()), + tool_calls: None, + tool_call_id: None, + } + } + + #[test] + fn plan_breakpoints_empty_when_disabled() { + let msgs = vec![cm("system", Some("sys")), cm("user", Some("hi"))]; + assert!(plan_cache_breakpoints(&msgs, false).is_empty()); + assert!(plan_cache_breakpoints(&[], true).is_empty()); + } + + #[test] + fn plan_breakpoints_system_and_user_on_first_turn() { + // [system, user] -> system(0) + rolling on the latest cacheable (user=1). + let msgs = vec![cm("system", Some("sys")), cm("user", Some("do it"))]; + assert_eq!(plan_cache_breakpoints(&msgs, true), vec![0, 1]); + } + + #[test] + fn plan_breakpoints_places_system_and_two_rolling_within_cap() { + // system, user, assistant(answer), tool(result), user(followup) + let msgs = vec![ + cm("system", Some("sys")), // 0 system + cm("user", Some("first")), // 1 + cm("assistant", Some("answer")), // 2 + cm("tool", Some("result")), // 3 + cm("user", Some("second")), // 4 latest user + ]; + let bps = plan_cache_breakpoints(&msgs, true); + // system(0) + rolling latest(4) + rolling one-turn-back before last user + // (the last cacheable index < 4, which is the tool result at 3). + assert_eq!(bps, vec![0, 3, 4]); + assert!(bps.len() <= MAX_CACHE_BREAKPOINTS); + } + + #[test] + fn plan_breakpoints_skips_messages_without_text_content() { + // assistant tool-call message (no content) must not be chosen. + let mut tool_call_msg = cm("assistant", None); + tool_call_msg.tool_calls = Some(vec![ToolCall { + id: "c1".to_string(), + type_: "function".to_string(), + function: FunctionCall { + name: "read".to_string(), + arguments: "{}".to_string(), + }, + }]); + let msgs = vec![ + cm("system", Some("sys")), + cm("user", Some("hi")), + tool_call_msg, + ]; + let bps = plan_cache_breakpoints(&msgs, true); + // index 2 (no content) is never marked. + assert!(!bps.contains(&2)); + assert!(bps.contains(&0)); + } + + #[test] + fn plan_breakpoints_never_exceed_four() { + let mut msgs = vec![cm("system", Some("sys"))]; + for i in 0..20 { + msgs.push(cm("user", Some(&format!("u{i}")))); + msgs.push(cm("assistant", Some(&format!("a{i}")))); + } + let bps = plan_cache_breakpoints(&msgs, true); + assert!(bps.len() <= MAX_CACHE_BREAKPOINTS); + // Sorted + deduped. + let mut sorted = bps.clone(); + sorted.sort_unstable(); + sorted.dedup(); + assert_eq!(bps, sorted); + } + #[test] fn validates_valid_json_against_schema() { let schema = json!({