liter-llm 1.12.2

//! 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;

// ~keep task_local! values must be 'static; Cell gives per-task mutability without Sync.
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));
}

/// 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,
        }
    }
}

/// 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,
    },
}

#[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)
    }
}

/// 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,
}

/// 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))
    }
}

/// 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 {
            return None;
        }
        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;
        }
        Some(entry.response.clone())
    }

    /// Remove an expired entry (eviction under write lock).
    fn remove_expired(&mut self, key: u64) {
        let ttl = self.ttl;
        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);
        }
    }

    fn insert(&mut self, key: u64, request_body: String, response: CachedResponse) {
        if self.map.contains_key(&key) {
            self.order.retain(|k| *k != key);
        }

        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 + '_>> {
        // ~keep Perform synchronous cache read/expiry/hit-count work under one write lock to avoid TOCTOU.
        // ~keep Sharded locks are the upgrade path if this single-lock path becomes contentious.
        let hit = self.inner.write().ok().and_then(|mut cache| {
            // ~keep Check validity first; `get_if_valid` handles expiry inline.
            let hit = cache.get_if_valid(key, request_body);
            if hit.is_none() {
                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))
    }
}

/// 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(),
        }
    }
}

/// 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);
            if self.store.get(key, &body).await.is_none() {
                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()?,
        _ => 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)> {
    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()?;
            let params = serde_json::json!({
                "temperature": r.temperature,
                "top_p": r.top_p,
                "max_tokens": r.max_tokens,
                "n": r.n,
                "stop": r.stop,
            });
            let tenant_id: Option<String> = req_tenant.or_else(|| {
                r.user
                    .as_deref()
                    .and_then(|u| u.strip_prefix("tenant:"))
                    .map(str::to_owned)
            });
            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 {
        // ~keep SAFETY: StandardCachePolicy only reads metadata, so the static empty map is shared safely.
        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);

        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();

        // ~keep Consume the poll-ready service instance and leave a fresh standby clone.
        let standby = self.inner.clone();
        let mut inner = std::mem::replace(&mut self.inner, standby);
        let fut = inner.call(req);

        Box::pin(async move {
            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");
            }

            if decision.use_semantic
                && let (Some(ep), Some(vs)) = (&embedding_provider, &vector_store)
                && let Some((_, ref body)) = key_and_body
            {
                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()?;
                    // ~keep Semantic hits must use the stored original body for collision checking.
                    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");
            }

            record_cache_state(CacheState::Miss);
            let resp = fut.await?;

            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 {
                    store.put(k, body.clone(), cached_resp).await;
                    if let Some(ttl) = decision.ttl_override {
                        store.set_ttl(k, ttl).await;
                    }

                    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,
                            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)
        })
    }
}

#[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);

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

        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);

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

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

        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"
        );
    }

    #[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);
        store
            .put(
                1,
                "body".into(),
                CachedResponse::Chat(crate::tower::tests_common::make_chat_response("gpt-4")),
            )
            .await;
        store.set_ttl(1, Duration::from_nanos(1)).await;
        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;
        let _ = store.get(42, "req").await;
        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");
    }

    #[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);

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

        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));

        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;

        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);

        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"
        );
    }

    /// 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);

        let mut req_a = chat_req("gpt-4");
        req_a.user = Some("tenant:acme".into());

        let mut req_b = chat_req("gpt-4");
        req_b.user = Some("tenant:globex".into());

        svc.call(LlmRequest::Chat(req_a.clone())).await.unwrap();
        assert_eq!(call_count.load(Ordering::SeqCst), 1, "first call must miss");

        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"
        );

        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);

        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,
            }),
        ];

        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,
            }),
        ];

        svc.call(LlmRequest::Chat(req_a.clone())).await.unwrap();
        assert_eq!(call_count.load(Ordering::SeqCst), 1, "first call must 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"
        );

        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");
    }

    #[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;

        assert_eq!(call_count.load(Ordering::SeqCst), 0, "warm must not call inner service");
    }

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

        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);

        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;

        #[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,
            ttl: Duration::from_nanos(1),
            backend: CacheBackend::default(),
        };
        let mut svc = CacheLayer::new(config).layer(inner);

        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");

        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);

        for _ in 0..1000 {
            let _ = svc.call(LlmRequest::Chat(chat_req("gpt-4"))).await;
        }
    }
}