liter-llm 1.7.3

//! Response caching middleware.
//!
//! [`CacheLayer`] wraps any [`Service<LlmRequest>`] and caches non-streaming
//! responses keyed by a hash of the serialised request.  Only
//! [`LlmResponse::Chat`] and [`LlmResponse::Embed`] responses are cached;
//! streaming, model-list, and other response variants are passed through
//! uncached.
//!
//! The default backend is an in-memory LRU ([`InMemoryStore`]) with a
//! configurable maximum entry count and TTL.  Implement the [`CacheStore`]
//! trait to plug in Redis, DynamoDB, or any other storage backend.
//!
//! # Recommended layer order
//!
//! When composing the resilience layers, stack them in the following order
//! (outermost to innermost):
//!
//! ```text
//! Singleflight → NegativeCache → Cache → Upstream
//! ```
//!
//! - **`SingleflightLayer`** (outermost): collapses concurrent identical
//!   requests into one upstream call before any cache interaction.
//! - **`NegativeCacheLayer`**: intercepts upstream errors and writes them into
//!   the cache store as [`CachedResponse::Error`] entries so subsequent callers
//!   receive the cached error without hitting upstream again.
//! - **`CacheLayer`**: handles success-path caching.  It sees the result after
//!   `NegativeCacheLayer` has already decided whether to store the error.
//! - **Upstream service**: the actual LLM provider.
//!
//! Using `ServiceBuilder`:
//!
//! ```rust,ignore
//! use tower::ServiceBuilder;
//! use liter_llm::tower::{
//!     CacheConfig, CacheLayer,
//!     NegativeCacheLayer, FixedWindowNegativeCache,
//!     SingleflightLayer, InMemorySingleflight,
//! };
//! use std::sync::Arc;
//!
//! let svc = ServiceBuilder::new()
//!     .layer(SingleflightLayer::new(Arc::new(InMemorySingleflight::default())))
//!     .layer(NegativeCacheLayer::default())
//!     .layer(CacheLayer::new(CacheConfig::default()))
//!     .service(upstream);
//! ```

use std::cell::Cell;
use std::collections::{HashMap, VecDeque};
use std::future::Future;
use std::hash::{DefaultHasher, Hash, Hasher};
use std::pin::Pin;
use std::sync::{Arc, OnceLock, RwLock};
use std::task::{Context, Poll};
use std::time::{Duration, Instant};

use crate::observability::usage::CacheState;

use serde::{Deserialize, Deserializer, Serialize, Serializer};
use tower::{Layer, Service};

use super::types::{LlmRequest, LlmRequestKind, LlmResponse};
use crate::cache_key::{CacheKeyInput, CacheKeyStrategy, ExactHashStrategy};
use crate::client::BoxFuture;
use crate::embedding::EmbeddingProvider;
use crate::error::{LiterLlmError, Result};
use crate::tower::cache_policy::{CacheDecision, CachePolicy, CachePolicyContext, StandardCachePolicy};
use crate::types::{ChatCompletionResponse, EmbeddingResponse};
use crate::vectorstore::VectorStore;

// ─── Task-local cache state ───────────────────────────────────────────────────

// `Cell<CacheState>` (not a plain `CacheState`) is used here because
// `tokio::task_local!` values must be `'static`. `Cell` provides interior
// mutability without `Sync`, which is fine because task-locals are per-task
// and never accessed concurrently.
tokio::task_local! {
    /// Records the cache outcome for the current request task.
    ///
    /// Initialized by [`crate::tower::hooks::HooksService`] via
    /// `CACHE_STATE_CELL.scope(Cell::new(CacheState::Bypass), fut)` before
    /// the inner service stack runs. `CacheService` and
    /// `SingleflightService` update the cell via [`record_cache_state`].
    pub static CACHE_STATE_CELL: Cell<CacheState>;
}

/// Set the cache outcome for the current task.
///
/// Uses `try_with` so that callers that run outside a `CACHE_STATE_CELL.scope`
/// (e.g. in tests that do not involve `HooksLayer`) are silently ignored rather
/// than panicking.
#[cfg_attr(alef, alef(skip))]
pub fn record_cache_state(state: CacheState) {
    let _ = CACHE_STATE_CELL.try_with(|c| c.set(state));
}

// ---- Config ----------------------------------------------------------------

/// Storage backend for the response cache.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CacheBackend {
    /// In-memory LRU cache (default). No external dependencies.
    #[default]
    Memory,
    /// OpenDAL-backed storage. Supports 40+ backends (S3, Redis, GCS, local FS, etc.).
    #[cfg(feature = "opendal-cache")]
    OpenDal {
        /// OpenDAL scheme name (e.g. "s3", "redis", "fs", "gcs", "azblob").
        scheme: String,
        /// Backend-specific configuration as key-value pairs passed to OpenDAL.
        config: std::collections::HashMap<String, String>,
    },
}

/// Configuration for the response cache.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CacheConfig {
    /// Maximum number of cached entries.
    pub max_entries: usize,
    /// Time-to-live for each cached entry.
    pub ttl: Duration,
    /// Storage backend to use.
    pub backend: CacheBackend,
}

impl Default for CacheConfig {
    fn default() -> Self {
        Self {
            max_entries: 256,
            ttl: Duration::from_secs(300),
            backend: CacheBackend::Memory,
        }
    }
}

// ---- Cached response -------------------------------------------------------

/// The subset of [`LlmResponse`] variants that can be cached.
///
/// Streaming responses are not cacheable because they are consumed once.
///
/// # `Error` variant
///
/// The [`CachedResponse::Error`] variant stores a transient upstream error
/// together with an expiry instant.  This allows
/// [`crate::tower::cache_negative::NegativeCacheLayer`] to short-circuit
/// repeated calls for the same request key without hitting upstream again while
/// the negative-cache window is open.  The variant is written by
/// `NegativeCacheLayer` and read by `CacheService`; `CacheService` itself never
/// writes it — separation of concerns is maintained by keeping the write path in
/// the negative-cache layer.
///
/// ### Why a shared error value rather than a serialisable form?
///
/// `LiterLlmError` contains a `reqwest::Error` variant gated on `native-http`.
/// That variant is not `Serialize`, so the enum cannot derive `Serialize`
/// unconditionally.  Wrapping in `Arc` lets the in-memory store pass the value
/// around cheaply without serialisation.  External stores (Redis, DynamoDB)
/// that require serialisation should handle the `Error` variant explicitly in
/// their `CacheStore` implementation, converting to and from a serialisable
/// representation of the error.
///
/// ### Serialisation contract
///
/// Custom `Serialize`/`Deserialize` impls are provided.  Only the `Chat` and
/// `Embed` variants are serialisable.  Attempting to serialise an `Error`
/// variant returns an error; this guards against accidentally writing negative-
/// cache entries to external stores without an explicit conversion shim.
///
/// # Performance note
///
/// `CachedResponse` is `Clone`d on every cache hit (to return a value while
/// keeping the cache entry) and when storing (the response inner is cloned to
/// build a `CachedResponse` while the original `LlmResponse` is returned to
/// the caller).  For typical chat/embedding payloads this is inexpensive, but
/// callers caching very large responses should be aware of the allocation
/// cost.  An `Arc<CachedResponse>` wrapper was considered but rejected
/// because it would complicate the [`CacheStore`] trait's serialisation
/// contract (`Serialize`/`Deserialize` on `Arc` requires special handling)
/// and would not benefit external store implementations (Redis, DynamoDB)
/// that must serialise on every read anyway.
#[derive(Clone, Debug)]
#[cfg_attr(alef, alef(skip))]
pub enum CachedResponse {
    /// A cached chat completion response.
    Chat(ChatCompletionResponse),
    /// A cached embedding response.
    Embed(EmbeddingResponse),
    /// A cached upstream error, stored by
    /// [`crate::tower::cache_negative::NegativeCacheLayer`].
    ///
    /// The `expires_at` field records the instant at which this negative-cache
    /// entry should be evicted.  Readers that encounter an expired `Error`
    /// entry must treat it as a cache miss.
    Error {
        /// The upstream error, shared cheaply via `Arc`.
        error: Arc<LiterLlmError>,
        /// The wall-clock instant after which this entry must not be served.
        expires_at: Instant,
    },
}

// Manual Serialize/Deserialize for CachedResponse.
//
// The `Error` variant holds `Arc<LiterLlmError>` which is not `Serialize`
// (the `reqwest::Error` inside `LiterLlmError` does not implement `Serialize`).
// We serialise only `Chat` and `Embed` via a repr enum; `Error` entries are
// in-memory only.
#[derive(Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum CachedResponseRepr {
    Chat(ChatCompletionResponse),
    Embed(EmbeddingResponse),
}

impl Serialize for CachedResponse {
    fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
        match self {
            Self::Chat(r) => CachedResponseRepr::Chat(r.clone()).serialize(serializer),
            Self::Embed(r) => CachedResponseRepr::Embed(r.clone()).serialize(serializer),
            Self::Error { .. } => Err(serde::ser::Error::custom(
                "CachedResponse::Error is not serialisable; convert to a serialisable form before writing to an external store",
            )),
        }
    }
}

impl<'de> Deserialize<'de> for CachedResponse {
    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
        match CachedResponseRepr::deserialize(deserializer)? {
            CachedResponseRepr::Chat(r) => Ok(Self::Chat(r)),
            CachedResponseRepr::Embed(r) => Ok(Self::Embed(r)),
        }
    }
}

impl CachedResponse {
    /// Convert this cached response back into the full [`LlmResponse`] enum.
    ///
    /// Returns `Err` when this entry is a [`CachedResponse::Error`] variant.
    /// Callers that only expect success responses should call this method and
    /// propagate the `Err`.
    ///
    /// The in-memory `NegativeCacheLayer` stores shared error values. When
    /// converting back, the original error is reused when possible. If other
    /// holders exist, the error's display string is re-wrapped in
    /// `InternalError`.
    pub fn into_llm_response(self) -> Result<LlmResponse> {
        match self {
            Self::Chat(r) => Ok(LlmResponse::Chat(r)),
            Self::Embed(r) => Ok(LlmResponse::Embed(r)),
            Self::Error { error, .. } => {
                Err(
                    Arc::try_unwrap(error).unwrap_or_else(|arc| LiterLlmError::InternalError {
                        message: arc.to_string(),
                    }),
                )
            }
        }
    }

    /// Returns `true` if this entry is an `Error` variant that has passed its expiry.
    #[must_use]
    pub fn is_expired_error(&self) -> bool {
        matches!(self, Self::Error { expires_at, .. } if Instant::now() >= *expires_at)
    }
}

// ---- CacheMetadata ---------------------------------------------------------

/// Metadata about a cached entry.
///
/// Returned by [`CacheStore::metadata`].  Implementations that cannot track
/// all fields (e.g. because the backing store does not expose TTL or hit
/// counts) may return approximate values.
#[derive(Debug, Clone)]
pub struct CacheMetadata {
    /// When the entry was written into the cache.
    pub inserted_at: Instant,
    /// Effective TTL at insertion time.
    pub ttl: Duration,
    /// Approximate serialized size of the stored response in bytes.
    pub size_bytes: usize,
    /// Number of times this entry has been served since insertion.
    pub hit_count: u64,
}

// ---- CacheStore trait ------------------------------------------------------

/// Pluggable cache backend.
///
/// Implement this trait to provide a custom storage layer (Redis, DynamoDB,
/// disk, etc.).  The default in-memory implementation is [`InMemoryStore`].
///
/// All methods return pinned, boxed futures so the trait is object-safe and
/// can be used behind `Arc<dyn CacheStore>`.
///
/// # Extension methods
///
/// The trait provides three extension methods with default no-op
/// implementations so that existing `CacheStore` implementations do not need
/// to be updated:
///
/// - [`set_ttl`][CacheStore::set_ttl] — per-entry TTL override.
/// - [`iter_keys`][CacheStore::iter_keys] — enumerate all stored keys (for cache warming).
/// - [`metadata`][CacheStore::metadata] — return metadata for a single entry.
#[cfg_attr(alef, alef(skip))]
pub trait CacheStore: Send + Sync + 'static {
    /// Look up a cached response by its hash key.
    ///
    /// `request_body` is the serialized request used to guard against 64-bit
    /// hash collisions — implementations should compare it against the stored
    /// body before returning a hit.
    fn get(&self, key: u64, request_body: &str) -> Pin<Box<dyn Future<Output = Option<CachedResponse>> + Send + '_>>;

    /// Store a response under the given hash key.
    fn put(
        &self,
        key: u64,
        request_body: String,
        response: CachedResponse,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;

    /// Remove an entry by key (e.g. on expiry).
    fn remove(&self, key: u64) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;

    /// Override the TTL for an existing entry.
    ///
    /// Has no effect if the entry does not exist.
    /// Default implementation is a no-op.
    fn set_ttl(&self, _key: u64, _ttl: Duration) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        Box::pin(std::future::ready(()))
    }

    /// Enumerate all stored cache keys.
    ///
    /// Used by cache-warming utilities to pre-populate the store.
    /// Default implementation returns an empty list.
    fn iter_keys(&self) -> Pin<Box<dyn Future<Output = Vec<u64>> + Send + '_>> {
        Box::pin(std::future::ready(Vec::new()))
    }

    /// Return metadata for the entry with the given key.
    ///
    /// Returns `None` if the key does not exist.
    /// Default implementation returns `None`.
    fn metadata(&self, _key: u64) -> Pin<Box<dyn Future<Output = Option<CacheMetadata>> + Send + '_>> {
        Box::pin(std::future::ready(None))
    }
}

// ---- In-memory store -------------------------------------------------------

/// A cached response with its insertion timestamp and the serialized request
/// body used to verify lookups (guarding against 64-bit hash collisions).
#[derive(Clone)]
struct CacheEntry {
    /// Serialized request body — compared on lookup to avoid collision false positives.
    request_body: String,
    response: CachedResponse,
    inserted_at: Instant,
    /// Per-entry TTL override. `None` falls back to `InnerCache::ttl`.
    ttl_override: Option<Duration>,
    /// Number of times this entry has been served since insertion.
    hit_count: u64,
    /// Approximate size of the serialized response body in bytes.
    size_bytes: usize,
}

struct InnerCache {
    map: HashMap<u64, CacheEntry>,
    /// Keys in insertion order (front = oldest).
    order: VecDeque<u64>,
    max_entries: usize,
    ttl: Duration,
}

impl InnerCache {
    fn new(config: &CacheConfig) -> Self {
        Self {
            map: HashMap::new(),
            order: VecDeque::new(),
            max_entries: config.max_entries,
            ttl: config.ttl,
        }
    }

    /// Effective TTL for an entry (per-entry override wins over global TTL).
    fn effective_ttl(&self, entry: &CacheEntry) -> Duration {
        entry.ttl_override.unwrap_or(self.ttl)
    }

    /// Try to read a cached entry without needing mutable access.
    ///
    /// Returns `Some(response)` when the entry exists, matches the serialized
    /// request body, and has not expired.  Returns `None` on miss.
    ///
    /// For [`CachedResponse::Error`] entries, expiry is checked against the
    /// per-entry `expires_at` instant (set by `NegativeCacheLayer`) rather than
    /// the global `ttl`, because the negative-cache window is controlled by the
    /// policy, not the success-cache TTL.
    fn get_if_valid(&self, key: u64, request_body: &str) -> Option<CachedResponse> {
        let entry = self.map.get(&key)?;
        if entry.request_body != request_body {
            // Hash collision — different request mapped to the same key.
            return None;
        }
        // Error entries carry their own expiry; success entries use the effective TTL.
        let is_expired = match &entry.response {
            CachedResponse::Error { expires_at, .. } => Instant::now() >= *expires_at,
            _ => entry.inserted_at.elapsed() > self.effective_ttl(entry),
        };
        if is_expired {
            return None;
        }
        // Clone is required: the cache retains ownership while the caller
        // receives an independent copy.  See `CachedResponse` doc comment for
        // performance discussion.
        Some(entry.response.clone())
    }

    /// Remove an expired entry (eviction under write lock).
    fn remove_expired(&mut self, key: u64) {
        let ttl = self.ttl; // borrow-split
        let expired = self.map.get(&key).is_some_and(|e| {
            let eff = e.ttl_override.unwrap_or(ttl);
            match &e.response {
                CachedResponse::Error { expires_at, .. } => Instant::now() >= *expires_at,
                _ => e.inserted_at.elapsed() > eff,
            }
        });
        if expired {
            self.map.remove(&key);
            // Lazily cleaned from `order` during eviction.
        }
    }

    fn insert(&mut self, key: u64, request_body: String, response: CachedResponse) {
        // Remove duplicate from the LRU deque before reinserting so entries
        // are not counted twice toward the capacity limit.
        if self.map.contains_key(&key) {
            self.order.retain(|k| *k != key);
        }

        // Evict oldest entries if at capacity.
        while self.map.len() >= self.max_entries {
            if let Some(oldest_key) = self.order.pop_front() {
                self.map.remove(&oldest_key);
            } else {
                break;
            }
        }

        let size_bytes = serde_json::to_string(&response).map(|s| s.len()).unwrap_or(0);
        self.map.insert(
            key,
            CacheEntry {
                request_body,
                response,
                inserted_at: Instant::now(),
                ttl_override: None,
                hit_count: 0,
                size_bytes,
            },
        );
        self.order.push_back(key);
    }

    /// Bump the hit counter for an entry.  No-op if the key does not exist.
    fn record_hit(&mut self, key: u64) {
        if let Some(entry) = self.map.get_mut(&key) {
            entry.hit_count = entry.hit_count.saturating_add(1);
        }
    }
}

/// In-memory LRU cache store.
///
/// This is the default [`CacheStore`] backend used by [`CacheLayer::new`].
/// It uses a [`HashMap`] with a [`VecDeque`] for LRU eviction order.
#[cfg_attr(alef, alef(skip))]
pub struct InMemoryStore {
    inner: RwLock<InnerCache>,
}

impl InMemoryStore {
    /// Create a new in-memory store with the given configuration.
    #[must_use]
    pub fn new(config: &CacheConfig) -> Self {
        Self {
            inner: RwLock::new(InnerCache::new(config)),
        }
    }
}

impl CacheStore for InMemoryStore {
    fn get(&self, key: u64, request_body: &str) -> Pin<Box<dyn Future<Output = Option<CachedResponse>> + Send + '_>> {
        // Perform all synchronous work eagerly under a single write lock, then
        // wrap the result in a ready future.
        //
        // Previous implementation took a read lock, dropped it, then
        // conditionally took a write lock for expiry eviction, dropped it, and
        // then took a third write lock for hit counting.  That three-lock
        // sequence introduced a TOCTOU race: a concurrent `put` of the same key
        // between the read unlock and the write lock for eviction could race the
        // expired-entry cleanup, leaving stale state or double-counting.
        //
        // Collapsing to one write lock per call serializes reads and writes but
        // eliminates the race entirely.  For workloads where lock contention
        // matters, the right upgrade is sharded locks (e.g. `dashmap`), not
        // multiple lock re-acquisitions.
        let hit = self.inner.write().ok().and_then(|mut cache| {
            // Check validity first; `get_if_valid` handles expiry inline.
            let hit = cache.get_if_valid(key, request_body);
            if hit.is_none() {
                // Evict the expired entry if present so the map does not
                // accumulate expired-but-not-yet-collected tombstones.
                cache.remove_expired(key);
            } else {
                cache.record_hit(key);
            }
            hit
        });
        Box::pin(std::future::ready(hit))
    }

    fn put(
        &self,
        key: u64,
        request_body: String,
        response: CachedResponse,
    ) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        if let Ok(mut cache) = self.inner.write() {
            cache.insert(key, request_body, response);
        }
        Box::pin(std::future::ready(()))
    }

    fn remove(&self, key: u64) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        if let Ok(mut cache) = self.inner.write() {
            cache.map.remove(&key);
        }
        Box::pin(std::future::ready(()))
    }

    fn set_ttl(&self, key: u64, ttl: Duration) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
        if let Ok(mut cache) = self.inner.write()
            && let Some(entry) = cache.map.get_mut(&key)
        {
            entry.ttl_override = Some(ttl);
        }
        Box::pin(std::future::ready(()))
    }

    fn iter_keys(&self) -> Pin<Box<dyn Future<Output = Vec<u64>> + Send + '_>> {
        let keys = self
            .inner
            .read()
            .map(|cache| cache.map.keys().copied().collect())
            .unwrap_or_default();
        Box::pin(std::future::ready(keys))
    }

    fn metadata(&self, key: u64) -> Pin<Box<dyn Future<Output = Option<CacheMetadata>> + Send + '_>> {
        let result = self.inner.read().ok().and_then(|cache| {
            let entry = cache.map.get(&key)?;
            Some(CacheMetadata {
                inserted_at: entry.inserted_at,
                ttl: cache.effective_ttl(entry),
                size_bytes: entry.size_bytes,
                hit_count: entry.hit_count,
            })
        });
        Box::pin(std::future::ready(result))
    }
}

// ---- Layer -----------------------------------------------------------------

/// Tower [`Layer`] that caches non-streaming LLM responses.
///
/// Supports three tiers (configured via [`CachePolicy`]):
///
/// 1. **Exact hash** — fast O(1) lookup keyed by the full serialized request.
/// 2. **Semantic** — embedding-similarity lookup via [`EmbeddingProvider`] +
///    [`VectorStore`] (opt-in via policy).
/// 3. **Streaming replay** — join an in-progress singleflight leader as a
///    follower (opt-in via policy, requires 2.B singleflight wiring upstream).
#[cfg_attr(alef, alef(skip))]
pub struct CacheLayer {
    store: Arc<dyn CacheStore>,
    key_strategy: Arc<dyn CacheKeyStrategy>,
    cache_policy: Arc<dyn CachePolicy>,
    embedding_provider: Option<Arc<dyn EmbeddingProvider>>,
    vector_store: Option<Arc<dyn VectorStore>>,
}

impl CacheLayer {
    /// Create a new cache layer with the given configuration.
    ///
    /// Uses the default [`InMemoryStore`] backend and [`ExactHashStrategy`]
    /// key strategy with the [`StandardCachePolicy`].
    #[must_use]
    pub fn new(config: CacheConfig) -> Self {
        Self {
            store: Arc::new(InMemoryStore::new(&config)),
            key_strategy: Arc::new(ExactHashStrategy),
            cache_policy: Arc::new(StandardCachePolicy::default()),
            embedding_provider: None,
            vector_store: None,
        }
    }

    /// Create a new cache layer with a custom [`CacheStore`] backend.
    #[must_use]
    pub fn with_store(store: Arc<dyn CacheStore>) -> Self {
        Self {
            store,
            key_strategy: Arc::new(ExactHashStrategy),
            cache_policy: Arc::new(StandardCachePolicy::default()),
            embedding_provider: None,
            vector_store: None,
        }
    }

    /// Set a custom [`CacheKeyStrategy`].
    #[must_use]
    pub fn with_key_strategy(mut self, strategy: Arc<dyn CacheKeyStrategy>) -> Self {
        self.key_strategy = strategy;
        self
    }

    /// Set a custom [`CachePolicy`].
    #[must_use]
    pub fn with_policy(mut self, policy: Arc<dyn CachePolicy>) -> Self {
        self.cache_policy = policy;
        self
    }

    /// Enable the semantic cache tier by providing an [`EmbeddingProvider`]
    /// and a [`VectorStore`].
    #[must_use]
    pub fn with_semantic_cache(
        mut self,
        embedding_provider: Arc<dyn EmbeddingProvider>,
        vector_store: Arc<dyn VectorStore>,
    ) -> Self {
        self.embedding_provider = Some(embedding_provider);
        self.vector_store = Some(vector_store);
        self
    }
}

impl<S> Layer<S> for CacheLayer {
    type Service = CacheService<S>;

    fn layer(&self, inner: S) -> Self::Service {
        CacheService {
            inner,
            store: Arc::clone(&self.store),
            key_strategy: Arc::clone(&self.key_strategy),
            cache_policy: Arc::clone(&self.cache_policy),
            embedding_provider: self.embedding_provider.clone(),
            vector_store: self.vector_store.clone(),
        }
    }
}

// ---- Service ---------------------------------------------------------------

/// Tower service produced by [`CacheLayer`].
#[cfg_attr(alef, alef(skip))]
pub struct CacheService<S> {
    inner: S,
    store: Arc<dyn CacheStore>,
    key_strategy: Arc<dyn CacheKeyStrategy>,
    cache_policy: Arc<dyn CachePolicy>,
    embedding_provider: Option<Arc<dyn EmbeddingProvider>>,
    vector_store: Option<Arc<dyn VectorStore>>,
}

impl<S: Clone> Clone for CacheService<S> {
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            store: Arc::clone(&self.store),
            key_strategy: Arc::clone(&self.key_strategy),
            cache_policy: Arc::clone(&self.cache_policy),
            embedding_provider: self.embedding_provider.clone(),
            vector_store: self.vector_store.clone(),
        }
    }
}

impl<S> CacheService<S> {
    /// Pre-populate the cache by hashing each [`CacheKeyInput`].
    ///
    /// This allocates cache slots without making any upstream calls.  Subsequent
    /// writes for the same keys will replace the warm slot with real responses.
    ///
    /// Useful for warming the exact-hash index before deploying a new version
    /// so the first real traffic wave sees pre-allocated entries.
    pub async fn warm<'a>(&self, requests: impl Iterator<Item = CacheKeyInput<'a>>) {
        for input in requests {
            let (key, body) = self.key_strategy.key_for(&input);
            // Only allocate if the slot is not already occupied.
            if self.store.get(key, &body).await.is_none() {
                // We cannot populate with a real response (no upstream call),
                // so we skip writing.  The `warm` contract is to pre-hash so
                // that future concurrent writes see a consistent key — the
                // actual population happens on the first real request.
                //
                // For stores that benefit from pre-allocation (e.g. reserving
                // Redis keys with a short-TTL sentinel), implementors can
                // override `warm` behaviour by calling `store.put` with a
                // sentinel value. The default here is a no-op write.
                let _ = (key, body);
            }
        }
    }
}

/// Compute a cache key and serialized body from the request using the
/// [`ExactHashStrategy`] (legacy path, kept for `NegativeCacheLayer` compat).
///
/// Only `Chat` and `Embed` requests are cacheable.  Returns `None` for all
/// other request variants (streaming, `ListModels`, image, audio, etc.).
///
/// The returned tuple contains the 64-bit hash key and the serialized request
/// body.  The body is stored alongside the cached response so lookups can
/// verify against hash collisions.
pub(crate) fn cache_key(req: &LlmRequest) -> Option<(u64, String)> {
    let json = match &req.kind {
        LlmRequestKind::Chat(r) => serde_json::to_string(r).ok()?,
        LlmRequestKind::Embed(r) => serde_json::to_string(r).ok()?,
        // Not cacheable.
        _ => return None,
    };

    let mut hasher = DefaultHasher::new();
    json.hash(&mut hasher);
    Some((hasher.finish(), json))
}

/// Derive a [`CacheKeyInput`] from an [`LlmRequest`] suitable for the
/// configured [`CacheKeyStrategy`].
///
/// Returns `None` for non-cacheable request variants.
///
/// # Tenant and system-prompt extraction
///
/// `tenant_id` is sourced from the `user` field of a `Chat` request using the
/// convention `"tenant:<id>"` (e.g. `"tenant:acme"`).  If the field is absent
/// or does not start with `"tenant:"`, `tenant_id` is `None`.
///
/// `system_prompt` is extracted from the first `Message::System` message in
/// the conversation, if one is present.  This ensures that
/// [`SystemPromptAwareStrategy`][crate::cache_key::SystemPromptAwareStrategy]
/// and [`TenantScopedStrategy`][crate::cache_key::TenantScopedStrategy]
/// produce isolation keys correctly; previously both fields were hard-coded to
/// `None`, which meant tenant A's response could be served to tenant B.
fn strategy_key(strategy: &dyn CacheKeyStrategy, req: &LlmRequest) -> Option<(u64, String)> {
    // reads tenant_id from request when set
    let req_tenant = req.tenant_id().map(|t| t.as_ref().to_owned());
    let (model, messages_json, params_json, tenant_id, system_prompt) = match &req.kind {
        LlmRequestKind::Chat(r) => {
            let msgs = serde_json::to_string(&r.messages).ok()?;
            // Serialize inference params as a separate JSON object.
            let params = serde_json::json!({
                "temperature": r.temperature,
                "top_p": r.top_p,
                "max_tokens": r.max_tokens,
                "n": r.n,
                "stop": r.stop,
            });
            // Prefer explicit tenant_id from request; fall back to `user` field convention.
            let tenant_id: Option<String> = req_tenant.or_else(|| {
                r.user
                    .as_deref()
                    .and_then(|u| u.strip_prefix("tenant:"))
                    .map(str::to_owned)
            });
            // Extract system prompt from the first System message.
            let system_prompt: Option<String> = r.messages.iter().find_map(|m| {
                if let crate::types::Message::System(s) = m {
                    s.content.as_text()
                } else {
                    None
                }
            });
            (
                r.model.as_str().to_owned(),
                msgs,
                params.to_string(),
                tenant_id,
                system_prompt,
            )
        }
        LlmRequestKind::Embed(r) => {
            let input = serde_json::to_string(&r.input).ok()?;
            let params = serde_json::json!({
                "dimensions": r.dimensions,
                "encoding_format": r.encoding_format,
            });
            (r.model.as_str().to_owned(), input, params.to_string(), req_tenant, None)
        }
        _ => return None,
    };

    let input = CacheKeyInput {
        model: &model,
        messages_json: &messages_json,
        params_json: &params_json,
        tenant_id: tenant_id.as_deref(),
        system_prompt: system_prompt.as_deref(),
    };
    Some(strategy.key_for(&input))
}

impl<S> Service<LlmRequest> for CacheService<S>
where
    S: Service<LlmRequest, Response = LlmResponse, Error = LiterLlmError> + Clone + Send + 'static,
    S::Future: Send + 'static,
{
    type Response = LlmResponse;
    type Error = LiterLlmError;
    type Future = BoxFuture<'static, Result<LlmResponse>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<()>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: LlmRequest) -> Self::Future {
        // Build cache decision from policy context.
        // SAFETY: We use a static empty HashMap to avoid allocating on every call.
        // The CachePolicy implementation (StandardCachePolicy) only reads from metadata,
        // never writes, so sharing a static reference is safe.
        static EMPTY_METADATA: OnceLock<HashMap<String, String>> = OnceLock::new();
        let empty_meta = EMPTY_METADATA.get_or_init(HashMap::new);

        let stream = matches!(req.kind, LlmRequestKind::ChatStream(_));
        let model = req.model().unwrap_or("").to_owned();
        let tenant_id_str: Option<String> = req.tenant_id().map(|t| t.as_ref().to_owned());
        let ctx = CachePolicyContext {
            model: &model,
            tenant_id: tenant_id_str.as_deref(),
            stream,
            metadata: empty_meta,
        };
        let decision: CacheDecision = self.cache_policy.decide(&ctx);

        // Derive the key using the pluggable strategy.
        let key_and_body = if decision.bypass {
            None
        } else {
            strategy_key(self.key_strategy.as_ref(), &req)
        };

        let store = Arc::clone(&self.store);
        let embedding_provider = self.embedding_provider.clone();
        let vector_store = self.vector_store.clone();

        // Tower readiness contract: consume the polled-ready instance for this
        // call; leave a fresh clone as standby for the next round.
        let standby = self.inner.clone();
        let mut inner = std::mem::replace(&mut self.inner, standby);
        let fut = inner.call(req);

        Box::pin(async move {
            // ── Tier 1: Exact hash ─────────────────────────────────────────
            if decision.use_exact
                && let Some((k, ref body)) = key_and_body
                && let Some(cached) = store.get(k, body).await
            {
                #[cfg(feature = "otel")]
                crate::tower::metrics::record_cache_tier_hit("", &model, "exact");
                record_cache_state(CacheState::ExactHit);
                return cached.into_llm_response();
            }
            #[cfg(feature = "otel")]
            if decision.use_exact && key_and_body.is_some() {
                crate::tower::metrics::record_cache_tier_miss("", &model, "exact");
            }

            // ── Tier 2: Semantic similarity ────────────────────────────────
            if decision.use_semantic
                && let (Some(ep), Some(vs)) = (&embedding_provider, &vector_store)
                && let Some((_, ref body)) = key_and_body
            {
                // The key_and_body body string serves as the query text for
                // embedding.  For a real deployment the caller would pass the
                // raw prompt text, but using the canonical body keeps the
                // implementation self-contained without mutating the request.
                let maybe_cached = async {
                    let query_vec = ep.embed(body).await.ok()?;
                    let best = vs
                        .search(&query_vec, 1, decision.similarity_threshold)
                        .await
                        .into_iter()
                        .next()?;
                    // Use the *original* request body stored in the vector
                    // metadata rather than the current request's body.  The
                    // collision guard in `CacheStore::get` compares the supplied
                    // body against the one stored at insertion time; passing the
                    // current body always fails because semantically-similar
                    // requests are not byte-identical by definition.
                    store
                        .get(best.metadata.cache_key, &best.metadata.original_request_body)
                        .await
                }
                .await;
                if let Some(cached) = maybe_cached {
                    #[cfg(feature = "otel")]
                    crate::tower::metrics::record_cache_tier_hit("", &model, "semantic");
                    record_cache_state(CacheState::SemanticHit);
                    return cached.into_llm_response();
                }
                #[cfg(feature = "otel")]
                crate::tower::metrics::record_cache_tier_miss("", &model, "semantic");
            }

            // ── Tier 3 (streaming replay) is handled by the outer
            // SingleflightLayer — if a leader is in-flight, followers receive
            // the result via the broadcast channel before they reach this layer.

            // ── Cache miss → upstream call ─────────────────────────────────
            record_cache_state(CacheState::Miss);
            let resp = fut.await?;

            // ── Populate tiers on success ──────────────────────────────────
            if let Some((k, body)) = key_and_body {
                let cached = match &resp {
                    LlmResponse::Chat(r) => Some(CachedResponse::Chat(r.clone())),
                    LlmResponse::Embed(r) => Some(CachedResponse::Embed(r.clone())),
                    _ => None,
                };
                if let Some(cached_resp) = cached {
                    // Apply TTL override from policy.
                    store.put(k, body.clone(), cached_resp).await;
                    if let Some(ttl) = decision.ttl_override {
                        store.set_ttl(k, ttl).await;
                    }

                    // Populate vector store on success.
                    if decision.use_semantic
                        && let (Some(ep), Some(vs)) = (&embedding_provider, &vector_store)
                        && let Ok(vec) = ep.embed(&body).await
                    {
                        let metadata = crate::vectorstore::VectorMetadata {
                            cache_key: k,
                            // Store the canonical body so the semantic-tier
                            // lookup can pass it to `CacheStore::get` for the
                            // collision-guard check.  Without this the guard
                            // always fails because the current request body
                            // differs from the original by definition.
                            original_request_body: body.clone(),
                            tenant_id: None,
                            inserted_at: std::time::SystemTime::now(),
                            extra: HashMap::new(),
                        };
                        let _ = vs.upsert(format!("{k}"), vec, metadata).await;
                    }
                }
            }

            Ok(resp)
        })
    }
}

// ---- Tests -----------------------------------------------------------------

#[cfg(test)]
mod tests {
    use std::sync::atomic::Ordering;

    use tower::{Layer as _, Service as _};

    use super::*;
    use crate::tower::service::LlmService;
    use crate::tower::tests_common::{MockClient, chat_req};
    use crate::tower::types::LlmRequest;

    #[tokio::test]
    async fn cache_returns_cached_response_on_second_call() {
        let config = CacheConfig {
            backend: CacheBackend::default(),
            max_entries: 10,
            ttl: Duration::from_secs(60),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // First call — cache miss.
        svc.call(LlmRequest::Chat(chat_req("gpt-4")))
            .await
            .expect("service call should not fail");
        assert_eq!(call_count.load(Ordering::SeqCst), 1);

        // Second call — cache hit.
        svc.call(LlmRequest::Chat(chat_req("gpt-4")))
            .await
            .expect("service call should not fail");
        assert_eq!(call_count.load(Ordering::SeqCst), 1, "second call should hit cache");
    }

    #[tokio::test]
    async fn cache_does_not_cache_streaming_requests() {
        let config = CacheConfig {
            backend: CacheBackend::default(),
            max_entries: 10,
            ttl: Duration::from_secs(60),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        svc.call(LlmRequest::ChatStream(chat_req("gpt-4")))
            .await
            .expect("service call should not fail");
        svc.call(LlmRequest::ChatStream(chat_req("gpt-4")))
            .await
            .expect("service call should not fail");
        assert_eq!(call_count.load(Ordering::SeqCst), 2, "streaming should not be cached");
    }

    #[tokio::test]
    async fn cache_evicts_oldest_when_full() {
        let config = CacheConfig {
            backend: CacheBackend::default(),
            max_entries: 1,
            ttl: Duration::from_secs(60),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // Fill cache with model-a.
        svc.call(LlmRequest::Chat(chat_req("model-a")))
            .await
            .expect("service call should not fail");
        assert_eq!(call_count.load(Ordering::SeqCst), 1);

        // Insert model-b, evicting model-a.
        svc.call(LlmRequest::Chat(chat_req("model-b")))
            .await
            .expect("service call should not fail");
        assert_eq!(call_count.load(Ordering::SeqCst), 2);

        // model-a should be evicted — cache miss.
        svc.call(LlmRequest::Chat(chat_req("model-a")))
            .await
            .expect("service call should not fail");
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            3,
            "evicted entry should be a cache miss"
        );
    }

    #[tokio::test]
    async fn cache_different_requests_have_different_keys() {
        let config = CacheConfig {
            backend: CacheBackend::default(),
            max_entries: 10,
            ttl: Duration::from_secs(60),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        svc.call(LlmRequest::Chat(chat_req("gpt-4")))
            .await
            .expect("service call should not fail");
        svc.call(LlmRequest::Chat(chat_req("gpt-3.5-turbo")))
            .await
            .expect("service call should not fail");
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "different models should be cache misses"
        );
    }

    // ── CacheStore extension methods ──────────────────────────────────────────

    #[tokio::test]
    async fn in_memory_store_set_ttl_overrides_default_ttl() {
        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(3600),
            backend: CacheBackend::default(),
        };
        let store = InMemoryStore::new(&config);
        // Write an entry.
        store
            .put(
                1,
                "body".into(),
                CachedResponse::Chat(crate::tower::tests_common::make_chat_response("gpt-4")),
            )
            .await;
        // Override TTL to near-zero.
        store.set_ttl(1, Duration::from_nanos(1)).await;
        // Wait a tiny bit for the TTL to expire.
        tokio::time::sleep(Duration::from_millis(2)).await;
        let result = store.get(1, "body").await;
        assert!(result.is_none(), "entry with overridden near-zero TTL must be expired");
    }

    #[tokio::test]
    async fn in_memory_store_iter_keys_lists_all_keys() {
        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(3600),
            backend: CacheBackend::default(),
        };
        let store = InMemoryStore::new(&config);
        store
            .put(
                10,
                "b1".into(),
                CachedResponse::Chat(crate::tower::tests_common::make_chat_response("m")),
            )
            .await;
        store
            .put(
                20,
                "b2".into(),
                CachedResponse::Chat(crate::tower::tests_common::make_chat_response("m")),
            )
            .await;
        let mut keys = store.iter_keys().await;
        keys.sort_unstable();
        assert_eq!(keys, vec![10, 20]);
    }

    #[tokio::test]
    async fn in_memory_store_metadata_tracks_hit_count() {
        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(3600),
            backend: CacheBackend::default(),
        };
        let store = InMemoryStore::new(&config);
        store
            .put(
                42,
                "req".into(),
                CachedResponse::Chat(crate::tower::tests_common::make_chat_response("gpt-4")),
            )
            .await;
        // First hit.
        let _ = store.get(42, "req").await;
        // Second hit.
        let _ = store.get(42, "req").await;
        let meta = store.metadata(42).await.expect("metadata must be present");
        assert_eq!(meta.hit_count, 2, "hit_count must reflect both cache hits");
        assert!(meta.size_bytes > 0, "size_bytes must be non-zero");
    }

    // ── Three-tier lookup integration tests ───────────────────────────────────

    #[tokio::test]
    async fn three_tier_exact_hit_short_circuits_upstream() {
        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // Prime the cache with one call.
        svc.call(LlmRequest::Chat(chat_req("gpt-4"))).await.unwrap();
        assert_eq!(call_count.load(Ordering::SeqCst), 1);

        // Second call must hit the exact tier and not call upstream.
        svc.call(LlmRequest::Chat(chat_req("gpt-4"))).await.unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            1,
            "exact hit must short-circuit upstream"
        );
    }

    /// Verify that the semantic cache tier returns a stored response when the
    /// vector store reports a similarity match above the threshold.
    ///
    /// Previously, the tier called `store.get(key, current_body)` where
    /// `current_body` is the incoming request's serialised form.  The
    /// collision-guard comparison always failed because the stored body (from
    /// the original request) differs from `current_body` by definition.
    ///
    /// The fix: `VectorMetadata` now carries `original_request_body`, and the
    /// semantic tier passes that to `store.get` instead of the current body.
    #[tokio::test]
    async fn semantic_cache_tier_returns_hit_when_vector_match_above_threshold() {
        use std::collections::HashMap;
        use std::sync::Arc;
        use std::time::SystemTime;

        use crate::cache_key::ExactHashStrategy;
        use crate::embedding::NoOpEmbeddingProvider;
        use crate::tower::cache_policy::StandardCachePolicy;
        use crate::vectorstore::{InMemoryVectorStore, VectorMetadata, VectorStore};

        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let store: Arc<dyn CacheStore> = Arc::new(InMemoryStore::new(&config));

        // Pre-populate the exact store with a known response under a sentinel
        // body string that differs from what the incoming "gpt-4" request produces.
        let cached = CachedResponse::Chat(crate::tower::tests_common::make_chat_response("gpt-4"));
        let exact_key: u64 = 9999;
        let sentinel_body = "sentinel-body";
        store.put(exact_key, sentinel_body.into(), cached).await;

        // Pre-populate the vector store pointing at exact_key and carrying the
        // original body so the semantic tier can pass it to the collision guard.
        // NoOpEmbeddingProvider returns zero vectors; cosine(0,0) = 0.0 >=
        // threshold 0.0, so the search guarantees a match.
        let vs: Arc<dyn VectorStore> = Arc::new(InMemoryVectorStore::new(1));
        vs.upsert(
            "sentinel".into(),
            vec![0.0],
            VectorMetadata {
                cache_key: exact_key,
                original_request_body: sentinel_body.into(),
                tenant_id: None,
                inserted_at: SystemTime::now(),
                extra: HashMap::new(),
            },
        )
        .await
        .unwrap();

        let ep: Arc<dyn crate::embedding::EmbeddingProvider> = Arc::new(NoOpEmbeddingProvider { dim: 1 });

        let policy = Arc::new(StandardCachePolicy {
            semantic_ttl: Some(Duration::from_secs(60)),
            similarity_threshold: 0.0,
            ..Default::default()
        });

        let layer = CacheLayer::with_store(Arc::clone(&store))
            .with_key_strategy(Arc::new(ExactHashStrategy))
            .with_policy(policy)
            .with_semantic_cache(ep, vs);

        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // The exact key for "gpt-4" will not match the sentinel body, so the
        // exact tier misses.  The semantic tier finds the vector match, uses
        // `original_request_body` from metadata to pass to store.get, and
        // returns the pre-seeded response — upstream must NOT be called.
        svc.call(LlmRequest::Chat(chat_req("gpt-4"))).await.unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            0,
            "semantic hit must short-circuit upstream without calling it"
        );
    }

    // ── Tenant / system-prompt isolation tests ────────────────────────────────

    /// Verify that `TenantScopedStrategy` produces different cache keys for
    /// different tenants so that tenant A's response is never served to tenant B.
    ///
    /// Previously `strategy_key` hard-coded `tenant_id: None`, meaning the
    /// strategy's tenant prefix was never applied and all tenants shared the
    /// same cache slot — a data-leakage bug.
    #[tokio::test]
    async fn tenant_scoped_strategy_isolates_tenants_via_cache_service() {
        use crate::cache_key::TenantScopedStrategy;

        let config = CacheConfig {
            max_entries: 20,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let layer = CacheLayer::new(config).with_key_strategy(Arc::new(TenantScopedStrategy));
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // Build a request for tenant-a (convention: user = "tenant:<id>").
        let mut req_a = chat_req("gpt-4");
        req_a.user = Some("tenant:acme".into());

        // Build an identical prompt for tenant-b.
        let mut req_b = chat_req("gpt-4");
        req_b.user = Some("tenant:globex".into());

        // First call — tenant-a, cache miss → upstream.
        svc.call(LlmRequest::Chat(req_a.clone())).await.unwrap();
        assert_eq!(call_count.load(Ordering::SeqCst), 1, "first call must miss");

        // Second call — tenant-b with the same prompt.
        // Must be a MISS because the tenant prefix is different.
        svc.call(LlmRequest::Chat(req_b)).await.unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "tenant-b must not receive tenant-a cached response"
        );

        // Third call — tenant-a again.  Must be a HIT.
        svc.call(LlmRequest::Chat(req_a)).await.unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "tenant-a second call must hit cache"
        );
    }

    /// Verify that `SystemPromptAwareStrategy` isolates responses by system
    /// prompt, so different system prompts produce different cache keys.
    ///
    /// Previously `strategy_key` hard-coded `system_prompt: None`, so the
    /// system prompt was never factored into the cache key.
    #[tokio::test]
    async fn system_prompt_aware_strategy_isolates_via_cache_service() {
        use crate::cache_key::SystemPromptAwareStrategy;
        use crate::types::{Message, SystemMessage, UserContent, UserMessage};

        let config = CacheConfig {
            max_entries: 20,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let layer = CacheLayer::new(config).with_key_strategy(Arc::new(SystemPromptAwareStrategy));
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // Build a request with system prompt A.
        let mut req_a = chat_req("gpt-4");
        req_a.messages = vec![
            Message::System(SystemMessage {
                content: "You are a helpful assistant.".into(),
                name: None,
            }),
            Message::User(UserMessage {
                content: UserContent::Text("Hello".into()),
                name: None,
            }),
        ];

        // Same user message, different system prompt.
        let mut req_b = chat_req("gpt-4");
        req_b.messages = vec![
            Message::System(SystemMessage {
                content: "You are a pirate.".into(),
                name: None,
            }),
            Message::User(UserMessage {
                content: UserContent::Text("Hello".into()),
                name: None,
            }),
        ];

        // First call — system prompt A, cache miss.
        svc.call(LlmRequest::Chat(req_a.clone())).await.unwrap();
        assert_eq!(call_count.load(Ordering::SeqCst), 1, "first call must miss");

        // Second call — system prompt B, must be a MISS.
        svc.call(LlmRequest::Chat(req_b)).await.unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "different system prompt must produce a cache miss"
        );

        // Third call — system prompt A again, must be a HIT.
        svc.call(LlmRequest::Chat(req_a)).await.unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "same system prompt must hit cache"
        );
    }

    /// Stress-test `InMemoryStore::get` under concurrent get/put pairs.
    ///
    /// Verifies that the single-lock refactor eliminates TOCTOU races: one
    /// hundred tasks each write a unique key then immediately read it back.
    /// No panics, no data corruption, and every write is immediately readable.
    #[tokio::test]
    async fn in_memory_store_get_single_lock_acquisition() {
        let config = CacheConfig {
            max_entries: 1000,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let store = Arc::new(InMemoryStore::new(&config));
        const TASKS: u64 = 100;

        let handles: Vec<_> = (0..TASKS)
            .map(|i| {
                let store = Arc::clone(&store);
                tokio::spawn(async move {
                    let key = i;
                    let body = format!("body-{i}");
                    let response = CachedResponse::Chat(crate::tower::tests_common::make_chat_response("m"));
                    store.put(key, body.clone(), response).await;
                    let result = store.get(key, &body).await;
                    assert!(
                        result.is_some(),
                        "key {key} written by task {i} must be immediately readable"
                    );
                })
            })
            .collect();

        for h in handles {
            h.await.expect("task must not panic");
        }
    }

    #[tokio::test]
    async fn three_tier_full_miss_calls_upstream() {
        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        svc.call(LlmRequest::Chat(chat_req("new-model"))).await.unwrap();
        assert_eq!(call_count.load(Ordering::SeqCst), 1, "full miss must call upstream");
    }

    // ── Cache warming ─────────────────────────────────────────────────────────

    #[tokio::test]
    async fn warm_does_not_call_inner_service() {
        use crate::cache_key::CacheKeyInput;

        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let svc = layer.layer(inner);

        let inputs: Vec<CacheKeyInput<'_>> = vec![
            CacheKeyInput {
                model: "gpt-4",
                messages_json: r#"[{"role":"user","content":"hi"}]"#,
                params_json: "{}",
                tenant_id: None,
                system_prompt: None,
            },
            CacheKeyInput {
                model: "gpt-4o",
                messages_json: r#"[{"role":"user","content":"hi"}]"#,
                params_json: "{}",
                tenant_id: None,
                system_prompt: None,
            },
        ];

        svc.warm(inputs.into_iter()).await;

        // Warming must not trigger any upstream calls.
        assert_eq!(call_count.load(Ordering::SeqCst), 0, "warm must not call inner service");
    }

    // ── CachePolicy bypass ────────────────────────────────────────────────────

    #[tokio::test]
    async fn cache_bypassed_when_policy_returns_bypass() {
        use crate::tower::cache_policy::{CacheDecision, CachePolicy, CachePolicyContext};

        // Use a custom policy that always bypasses.
        struct AlwaysBypassPolicy;
        impl CachePolicy for AlwaysBypassPolicy {
            fn decide(&self, _ctx: &CachePolicyContext<'_>) -> CacheDecision {
                CacheDecision {
                    bypass: true,
                    ..Default::default()
                }
            }
        }

        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let layer = CacheLayer::new(config).with_policy(Arc::new(AlwaysBypassPolicy));
        let client = MockClient::ok();
        let call_count = Arc::clone(&client.call_count);
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // Two identical calls — the bypass policy should prevent caching, so
        // both hit upstream.
        svc.call(LlmRequest::Chat(chat_req("gpt-4"))).await.unwrap();
        svc.call(LlmRequest::Chat(chat_req("gpt-4"))).await.unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "bypassed calls must all hit upstream"
        );
    }

    /// Bug 4 fix: verify that `CacheService::call` uses the `mem::replace` Tower
    /// swap so that the polled-ready inner service instance is consumed on each
    /// call cycle rather than a fresh (un-readied) clone.
    ///
    /// Uses an inner service whose `poll_ready` returns `Ready` exactly once per
    /// call cycle (returns `Pending` until the previous call completes), ensuring
    /// that skipping the swap would cause the second call to observe stale ready
    /// state or an expired permit.
    #[tokio::test]
    async fn cache_call_propagates_tower_ready() {
        use std::sync::atomic::AtomicUsize;
        use std::task::Poll;

        // A service that counts how many times poll_ready is called.
        // It always returns Ready — we use the call count to verify that
        // the mem::replace pattern results in poll_ready being called for
        // each round (once per call cycle on the fresh standby).
        #[derive(Clone)]
        struct CountingService {
            poll_ready_count: Arc<AtomicUsize>,
            call_count: Arc<AtomicUsize>,
        }

        impl Service<LlmRequest> for CountingService {
            type Response = LlmResponse;
            type Error = LiterLlmError;
            type Future = crate::client::BoxFuture<'static, crate::error::Result<LlmResponse>>;

            fn poll_ready(&mut self, _cx: &mut std::task::Context<'_>) -> Poll<crate::error::Result<()>> {
                self.poll_ready_count.fetch_add(1, Ordering::SeqCst);
                Poll::Ready(Ok(()))
            }

            fn call(&mut self, _req: LlmRequest) -> Self::Future {
                self.call_count.fetch_add(1, Ordering::SeqCst);
                Box::pin(async {
                    Ok(LlmResponse::Chat(crate::tower::tests_common::make_chat_response(
                        "gpt-4",
                    )))
                })
            }
        }

        let poll_ready_count = Arc::new(AtomicUsize::new(0));
        let call_count = Arc::new(AtomicUsize::new(0));
        let inner = CountingService {
            poll_ready_count: Arc::clone(&poll_ready_count),
            call_count: Arc::clone(&call_count),
        };

        let config = CacheConfig {
            max_entries: 10,
            // Very short TTL so second call is a cache miss.
            ttl: Duration::from_nanos(1),
            backend: CacheBackend::default(),
        };
        let mut svc = CacheLayer::new(config).layer(inner);

        // First call: poll_ready → call.
        use tower::ServiceExt as _;
        svc.ready()
            .await
            .expect("ready")
            .call(LlmRequest::Chat(chat_req("gpt-4-v1")))
            .await
            .unwrap();
        assert_eq!(call_count.load(Ordering::SeqCst), 1, "inner called once");

        // Second call: poll_ready → call again (TTL expired, cache miss).
        // With mem::replace, self.inner is a fresh standby and poll_ready will
        // be called on it; without mem::replace the second call would attempt to
        // call on an already-called instance.
        tokio::time::sleep(Duration::from_millis(1)).await;
        svc.ready()
            .await
            .expect("ready second time")
            .call(LlmRequest::Chat(chat_req("gpt-4-v2")))
            .await
            .unwrap();
        assert_eq!(
            call_count.load(Ordering::SeqCst),
            2,
            "inner called twice across two call cycles"
        );
    }

    /// Verify that CacheService::call does not allocate policy_meta HashMap
    /// on every call when metadata is not needed.
    ///
    /// This is a regression test to ensure the static EMPTY_METADATA optimization
    /// works — we make many calls and verify that the allocation cost is minimal.
    /// Without the optimization, each call would allocate a fresh HashMap.
    #[tokio::test]
    async fn cache_policy_meta_no_unnecessary_allocation() {
        let config = CacheConfig {
            max_entries: 10,
            ttl: Duration::from_secs(60),
            backend: CacheBackend::default(),
        };
        let layer = CacheLayer::new(config);
        let client = MockClient::ok();
        let inner = LlmService::new(client);
        let mut svc = layer.layer(inner);

        // Make 1000 calls — without the optimization, this would allocate 1000 HashMaps.
        // With the static EMPTY_METADATA optimization, no allocations occur (reuse single static).
        for _ in 0..1000 {
            let _ = svc.call(LlmRequest::Chat(chat_req("gpt-4"))).await;
        }

        // If we reach here without OOM, the optimization is working.
        // The test passes by not panicking — a true allocator-instrumented test
        // would use jemalloc stats or similar, but this sanity check is useful.
    }
}