feat: add opt-in exact-match caching for LLM responses

crates/adaptive/Cargo.toml

@@ -23,6 +23,7 @@ chrono = { version = "0.4", features = ["serde"] }
 thiserror = "2"
 tdigest = { version = "0.2", features = ["use_serde"] }
 tokio = { version = "1", features = ["rt", "macros", "sync", "time"] }
+tokio-stream = { version = "0.1", default-features = false, features = ["sync"] }
 regex = "1"
 serde_json_canonicalizer = "0.3"
 sha2 = "0.11"
@@ -35,7 +36,6 @@ redis-backend = ["redis"]
 
 [dev-dependencies]
 tokio = { version = "1", default-features = false, features = ["rt", "macros", "sync", "time", "test-util", "rt-multi-thread"] }
-tokio-stream = { version = "0.1", default-features = false }
 
 [[test]]
 name = "redis_integration"
@@ -48,3 +48,11 @@ path = "tests/integration/runtime_integration_tests.rs"
 [[test]]
 name = "acg_module_surface"
 path = "tests/integration/acg_module_surface_tests.rs"
+
+[[test]]
+name = "response_cache_integration"
+path = "tests/integration/response_cache_tests.rs"
+
+[[test]]
+name = "response_cache_benchmark_integration"
+path = "tests/integration/response_cache_benchmark_tests.rs"

crates/adaptive/src/config.rs

@@ -7,6 +7,8 @@ use nemo_relay::plugin::ConfigPolicy;
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value as Json};
 
+use crate::response_cache::config::{BackendConfig, KEY_STRATEGY_EXACT_REQUEST};
+
 /// Canonical config document for the adaptive plugin component.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AdaptiveConfig {
@@ -32,6 +34,10 @@ pub struct AdaptiveConfig {
     /// Adaptive Cache Governor settings.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub acg: Option<AcgComponentConfig>,
+    /// Opt-in LLM response cache (exact-match). When present, the
+    /// adaptive plugin installs the response-cache execution intercept(s).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub response_cache: Option<ResponseCacheConfig>,
     /// Adaptive-local unsupported-config policy.
     #[serde(default)]
     pub policy: ConfigPolicy,
@@ -47,6 +53,7 @@ impl Default for AdaptiveConfig {
             adaptive_hints: None,
             tool_parallelism: None,
             acg: None,
+            response_cache: None,
             policy: ConfigPolicy::default(),
         }
     }
@@ -184,6 +191,56 @@ impl Default for AcgComponentConfig {
     }
 }
 
+/// Configuration for the adaptive plugin's `response_cache` feature
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct ResponseCacheConfig {
+    /// How long a stored answer stays reusable, in seconds.
+    pub ttl_seconds: u64,
+    /// Required, non-empty trust-domain partition folded into every key.
+    ///
+    /// One response-cache namespace must not span mutually untrusted tenants
+    /// or upstreams. The empty default is an unconfigured sentinel rejected
+    /// when the response-cache section is enabled.
+    pub namespace: String,
+    /// Execution-intercept priority; lower runs first/outermost (default `50`).
+    pub priority: i32,
+    /// Probability in `[0.0, 1.0]` of skipping the cache and running live.
+    pub bypass_rate: f64,
+    /// Cache nondeterministic requests too. Set `false` to cache only
+    /// requests explicitly pinned deterministic (`temperature` = 0) — absent
+    /// or unreadable temperatures count as nondeterministic.
+    pub cache_nondeterministic: bool,
+    /// Key strategy. Only [`KEY_STRATEGY_EXACT_REQUEST`] is supported.
+    pub key_strategy: String,
+    /// Request headers (case-insensitive) folded into the key; never auth headers.
+    pub header_allowlist: Vec<String>,
+    /// Storage backend selection.
+    pub backend: BackendConfig,
+}
+
+impl Default for ResponseCacheConfig {
+    fn default() -> Self {
+        Self {
+            ttl_seconds: 3600,
+            namespace: String::new(),
+            priority: 50,
+            bypass_rate: 0.0,
+            cache_nondeterministic: false,
+            key_strategy: KEY_STRATEGY_EXACT_REQUEST.to_string(),
+            header_allowlist: Vec::new(),
+            backend: BackendConfig::default(),
+        }
+    }
+}
+
+impl ResponseCacheConfig {
+    /// TTL as a [`std::time::Duration`].
+    pub fn ttl(&self) -> std::time::Duration {
+        std::time::Duration::from_secs(self.ttl_seconds)
+    }
+}
+
 fn default_adaptive_config_version() -> u32 {
     1
 }
@@ -254,6 +311,13 @@ nemo_relay::editor_config! {
             nested: AcgComponentConfig,
             default: AcgComponentConfig,
         },
+        response_cache => {
+            label: "response_cache",
+            kind: Section,
+            optional: true,
+            nested: ResponseCacheConfig,
+            default: ResponseCacheConfig,
+        },
         policy => {
             label: "policy",
             kind: Section,
@@ -326,6 +390,24 @@ nemo_relay::editor_config! {
     }
 }
 
+nemo_relay::editor_config! {
+    impl ResponseCacheConfig {
+        ttl_seconds => { label: "ttl_seconds", kind: Integer },
+        namespace => { label: "namespace", kind: String },
+        priority => { label: "priority", kind: Integer },
+        bypass_rate => { label: "bypass_rate", kind: Float },
+        cache_nondeterministic => { label: "cache_nondeterministic", kind: Boolean },
+        key_strategy => { label: "key_strategy", kind: String },
+        header_allowlist => { label: "header_allowlist", kind: Json },
+        backend => {
+            label: "backend",
+            kind: Section,
+            nested: BackendConfig,
+            default: BackendConfig,
+        },
+    }
+}
+
 nemo_relay::editor_config! {
     impl crate::acg::stability::StabilityThresholds {
         stable_threshold => { label: "stable_threshold", kind: Float },

crates/adaptive/src/lib.rs

@@ -34,6 +34,8 @@ pub mod learner;
 pub mod plugin_component;
 #[cfg(feature = "redis-backend")]
 pub mod redis;
+/// Opt-in LLM response cache (exact-match).
+pub mod response_cache;
 mod runtime;
 /// Storage backends and backend traits for adaptive state persistence.
 pub mod storage;
@@ -45,8 +47,8 @@ pub mod trie;
 pub mod types;
 
 pub use config::{
-    AcgComponentConfig, AdaptiveConfig, AdaptiveHintsComponentConfig, BackendSpec, StateConfig,
-    TelemetryComponentConfig, ToolParallelismComponentConfig,
+    AcgComponentConfig, AdaptiveConfig, AdaptiveHintsComponentConfig, BackendSpec,
+    ResponseCacheConfig, StateConfig, TelemetryComponentConfig, ToolParallelismComponentConfig,
 };
 pub use context_helpers::{
     LATENCY_SENSITIVITY_POINTER, extract_scope_path, read_manual_latency_sensitivity,
@@ -55,6 +57,7 @@ pub use context_helpers::{
 pub use error::{AdaptiveError, Result};
 #[cfg(feature = "redis-backend")]
 pub use redis::RedisBackend;
+pub use response_cache::RESPONSE_CACHE_MARK;
 pub use runtime::features::AdaptiveRuntime;
 pub use storage::erased::AnyBackend;
 pub use storage::memory::InMemoryBackend;

crates/adaptive/src/plugin_component.rs

@@ -200,6 +200,7 @@ fn validate_adaptive_plugin_config_with_policy(
             "adaptive_hints",
             "tool_parallelism",
             "acg",
+            "response_cache",
             "policy",
         ],
     );
@@ -302,10 +303,74 @@ fn validate_adaptive_plugin_config_with_policy(
         );
     }
 
+    if let Some(response_cache_json) = plugin_config
+        .get("response_cache")
+        .and_then(Json::as_object)
+    {
+        validate_unknown_fields(
+            &mut diagnostics,
+            &config.policy,
+            Some("response_cache".to_string()),
+            response_cache_json,
+            &[
+                "ttl_seconds",
+                "namespace",
+                "priority",
+                "bypass_rate",
+                "cache_nondeterministic",
+                "key_strategy",
+                "header_allowlist",
+                "backend",
+            ],
+        );
+        if let Some(backend_json) = response_cache_json.get("backend").and_then(Json::as_object) {
+            validate_unknown_fields(
+                &mut diagnostics,
+                &config.policy,
+                Some("response_cache.backend".to_string()),
+                backend_json,
+                &["kind", "config"],
+            );
+            let backend_kind = backend_json
+                .get("kind")
+                .and_then(Json::as_str)
+                .unwrap_or("in_memory");
+            if let Some(backend_config_json) = backend_json.get("config").and_then(Json::as_object)
+            {
+                validate_response_cache_backend_config_fields(
+                    &mut diagnostics,
+                    &config.policy,
+                    backend_kind,
+                    backend_config_json,
+                );
+            }
+        }
+    }
+
     diagnostics.extend(AdaptiveRuntime::validate_config(&config).diagnostics);
     diagnostics
 }
 
+fn validate_response_cache_backend_config_fields(
+    diagnostics: &mut Vec<ConfigDiagnostic>,
+    policy: &ConfigPolicy,
+    backend_kind: &str,
+    backend_config: &Map<String, Json>,
+) {
+    let known_fields: &[&str] = match backend_kind {
+        "in_memory" => &["max_bytes"],
+        "redis" => &["url", "key_prefix"],
+        _ => return,
+    };
+    validate_unknown_fields(
+        diagnostics,
+        policy,
+        Some(format!("response_cache.backend.{backend_kind}")),
+        backend_config,
+        known_fields,
+    );
+}
+
 fn validate_backend_config_fields(
     diagnostics: &mut Vec<ConfigDiagnostic>,
     policy: &ConfigPolicy,

crates/adaptive/src/redis.rs

@@ -31,6 +31,57 @@ use crate::trie::serialization::TrieEnvelope;
 use crate::types::plan::ExecutionPlan;
 use crate::types::records::RunRecord;
 
+/// Connect to Redis, returning the client and an auto-reconnecting
+/// [`ConnectionManager`].
+///
+/// # Errors
+///
+/// Returns [`AdaptiveError::Storage`] if the client cannot be created or the
+/// connection cannot be established.
+pub(crate) async fn connect(url: &str) -> Result<(Client, ConnectionManager)> {
+    log::info!(
+        target: "nemo_relay.plugin",
+        event = "plugin_resource_access_pending",
+        plugin_kind = "adaptive",
+        resource_kind = "redis",
+        permission = "connect";
+        "Plugin Redis connectivity validation started"
+    );
+    let client = redis::Client::open(url).map_err(|e| {
+        log::warn!(
+            target: "nemo_relay.plugin",
+            event = "plugin_resource_access_failed",
+            plugin_kind = "adaptive",
+            resource_kind = "redis",
+            permission = "connect",
+            reason = "client_configuration";
+            "Plugin resource access validation failed"
+        );
+        AdaptiveError::Storage(format!("redis client: {e}"))
+    })?;
+    let conn = client.get_connection_manager().await.map_err(|e| {
+        log::warn!(
+            target: "nemo_relay.plugin",
+            event = "plugin_resource_access_failed",
+            plugin_kind = "adaptive",
+            resource_kind = "redis",
+            permission = "connect",
+            reason = "connection_failed";
+            "Plugin resource access validation failed"
+        );
+        AdaptiveError::Storage(format!("redis connection: {e}"))
+    })?;
+    log::info!(
+        target: "nemo_relay.plugin",
+        event = "plugin_resource_connected",
+        plugin_kind = "adaptive",
+        resource_kind = "redis",
+        permission = "connect";
+        "Plugin Redis connectivity established"
+    );
+    Ok((client, conn))
+}
+
 /// A Redis-backed storage backend for cross-process shared state.
 ///
 /// Uses [`ConnectionManager`] which is `Clone` (internally `Arc`-based) and
@@ -55,46 +106,7 @@ impl RedisBackend {
     /// Returns [`AdaptiveError::Storage`] if the client cannot be created or the
     /// connection cannot be established.
     pub async fn new(url: &str, key_prefix: impl Into<String>) -> Result<Self> {
-        log::info!(
-            target: "nemo_relay.plugin",
-            event = "plugin_resource_access_pending",
-            plugin_kind = "adaptive",
-            resource_kind = "redis",
-            permission = "connect";
-            "Plugin Redis connectivity validation started"
-        );
-        let client = redis::Client::open(url).map_err(|e| {
-            log::warn!(
-                target: "nemo_relay.plugin",
-                event = "plugin_resource_access_failed",
-                plugin_kind = "adaptive",
-                resource_kind = "redis",
-                permission = "connect",
-                reason = "client_configuration";
-                "Plugin resource access validation failed"
-            );
-            AdaptiveError::Storage(format!("redis client: {e}"))
-        })?;
-        let conn = client.get_connection_manager().await.map_err(|e| {
-            log::warn!(
-                target: "nemo_relay.plugin",
-                event = "plugin_resource_access_failed",
-                plugin_kind = "adaptive",
-                resource_kind = "redis",
-                permission = "connect",
-                reason = "connection_failed";
-                "Plugin resource access validation failed"
-            );
-            AdaptiveError::Storage(format!("redis connection: {e}"))
-        })?;
-        log::info!(
-            target: "nemo_relay.plugin",
-            event = "plugin_resource_connected",
-            plugin_kind = "adaptive",
-            resource_kind = "redis",
-            permission = "connect";
-            "Plugin Redis connectivity established"
-        );
+        let (client, conn) = connect(url).await?;
         Ok(Self {
             client,
             conn,