klieo-core 3.3.0

//! Per-stream SSE event buffer used by HTTP/SSE transports for
//! resumption via `Last-Event-ID` / `klieo/tools/resume`.
//!
//! The [`ResumeBuffer`] trait is opaque on `stream_id` (`klieo-a2a`
//! passes task IDs; `klieo-mcp-server` passes stringified
//! progressTokens). [`NoopResumeBuffer`] is the zero-cost default for
//! adopters who do not enable resumption. `KvResumeBuffer` (added in
//! T5) is the default `KvStore`-backed impl.

use async_trait::async_trait;
use bytes::Bytes;
use futures_core::Stream;
use std::pin::Pin;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use thiserror::Error;

/// Maximum events retained per stream before the oldest is evicted.
pub const RESUME_RING_CAP: usize = 256;

/// Idle TTL — buckets with no writes for this duration are evicted by
/// the sweeper.
pub const RESUME_IDLE_TTL: Duration = Duration::from_secs(5 * 60);

/// How often the sweeper walks buckets.
pub const RESUME_SWEEP_INTERVAL: Duration = Duration::from_secs(60);

/// Grace period between a stream being marked terminal and its bucket
/// being evicted by the sweeper.
pub const RESUME_TERMINAL_GRACE: Duration = Duration::from_secs(30);

/// Per-stream SSE event buffer. `stream_id` is opaque to the buffer —
/// `klieo-a2a` passes task IDs; `klieo-mcp-server` passes stringified
/// progressTokens.
#[async_trait]
pub trait ResumeBuffer: Send + Sync {
    /// Append one event under `stream_id` with the given `event_id`.
    /// Implementations evict the oldest event when the per-stream
    /// ring is full (cap = [`RESUME_RING_CAP`]).
    ///
    /// **Caller contract: `event_id` must be strictly monotonically
    /// increasing per `stream_id`.** Implementations are free to drop,
    /// reorder, or silently overwrite out-of-order writes; A2A's
    /// per-task counter and MCP's per-progressToken counter both
    /// satisfy this contract by construction.
    async fn record(
        &self,
        stream_id: &str,
        event_id: u64,
        payload: Bytes,
    ) -> Result<(), ResumeError>;

    /// Replay events with id > since_id, in monotonic order. Returns
    /// [`ResumeError::Expired`] if the cursor is before the oldest
    /// retained id; [`ResumeError::NotFound`] if no buffer exists for
    /// the stream.
    async fn replay(
        &self,
        stream_id: &str,
        since_id: u64,
    ) -> Result<Pin<Box<dyn Stream<Item = (u64, Bytes)> + Send>>, ResumeError>;

    /// Mark the stream terminal. Buffer is eligible for sweeper
    /// eviction after [`RESUME_TERMINAL_GRACE`].
    async fn close(&self, stream_id: &str) -> Result<(), ResumeError>;

    /// Return `true` if the stream has been marked terminal via `close`.
    /// Implementations that do not track terminal state return `false`
    /// (conservative: assume live). Used by `stream_resume` to decide
    /// whether to attach a live-tail subscription after replay drains.
    async fn is_terminal(&self, _stream_id: &str) -> Result<bool, ResumeError> {
        Ok(false)
    }
}

/// Failure modes for [`ResumeBuffer`].
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum ResumeError {
    /// The requested cursor is before the oldest retained id.
    #[error("resume window expired (oldest retained id > requested {since_id})")]
    Expired {
        /// Cursor the caller supplied.
        since_id: u64,
    },
    /// No buffer exists for this stream id.
    #[error("stream not found: {0}")]
    NotFound(String),
    /// Backend error (KV store failure, serialisation error, etc).
    #[error("buffer backend: {0}")]
    Backend(#[source] Box<dyn std::error::Error + Send + Sync>),
}

/// Zero-cost default — `record` is a no-op, `replay` returns
/// [`ResumeError::NotFound`], `close` is a no-op. Used by builders
/// when resumption is not enabled.
pub struct NoopResumeBuffer;

#[async_trait]
impl ResumeBuffer for NoopResumeBuffer {
    async fn record(
        &self,
        _stream_id: &str,
        _event_id: u64,
        _payload: Bytes,
    ) -> Result<(), ResumeError> {
        Ok(())
    }

    async fn replay(
        &self,
        stream_id: &str,
        _since_id: u64,
    ) -> Result<Pin<Box<dyn Stream<Item = (u64, Bytes)> + Send>>, ResumeError> {
        Err(ResumeError::NotFound(stream_id.to_string()))
    }

    async fn close(&self, _stream_id: &str) -> Result<(), ResumeError> {
        Ok(())
    }
}

use crate::bus::KvStore;
use crate::error::BusError;

/// Prefix for resume-buffer KV buckets: `klieo.resume.{stream_id}`.
const RESUME_BUCKET_PREFIX: &str = "klieo.resume.";
/// Directory bucket listing all active stream IDs for the sweeper.
const RESUME_DIRECTORY_BUCKET: &str = "klieo.resume.index";
const META_KEY: &str = "meta";
const EVENT_KEY_PREFIX: &str = "event/";

/// Sweeper interval tuning. Default values match [`RESUME_IDLE_TTL`],
/// [`RESUME_SWEEP_INTERVAL`], and [`RESUME_TERMINAL_GRACE`]. Tests
/// override with short values to exercise the sweeper without sleeping
/// minutes.
#[derive(Debug, Clone)]
pub struct SweeperTuning {
    /// Buckets with no writes for this duration are eligible for eviction.
    pub idle_ttl: Duration,
    /// How often the sweeper walks the directory bucket.
    pub sweep_interval: Duration,
    /// Grace period between a stream being marked terminal and its eviction.
    pub terminal_grace: Duration,
}

impl Default for SweeperTuning {
    fn default() -> Self {
        Self {
            idle_ttl: RESUME_IDLE_TTL,
            sweep_interval: RESUME_SWEEP_INTERVAL,
            terminal_grace: RESUME_TERMINAL_GRACE,
        }
    }
}

/// Summary of one MCP resume-buffer stream. Returned by
/// [`KvResumeBuffer::list_all_streams`] for operator visibility.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct StreamEntry {
    /// Opaque stream key — matches the `stream_id` passed to [`ResumeBuffer::record`].
    pub stream_id: String,
    /// `true` after [`ResumeBuffer::close`] has been called.
    pub is_terminal: bool,
    /// Next event id to be assigned — a proxy for total events recorded.
    pub next_id: u64,
    /// Unix milliseconds of the most recent [`ResumeBuffer::record`] call.
    pub last_write_unix_ms: u64,
}

/// One page of [`StreamEntry`] plus an opaque cursor for the next page.
/// Returned by [`KvResumeBuffer::list_streams_paged`].
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct StreamPage {
    /// Stream summaries in this page, in directory-key ascending order.
    pub entries: Vec<StreamEntry>,
    /// Cursor to pass as the next call's `cursor`. `None` once the final
    /// directory key has been returned (the scan is exhausted).
    pub next: Option<String>,
}

/// `KvStore`-backed [`ResumeBuffer`]. Layout per stream:
///
/// - `klieo.resume.{stream_id}` / `meta` → JSON `BufferMeta`.
/// - `klieo.resume.{stream_id}` / `event/{id}` → raw event payload bytes.
/// - `klieo.resume.index` / `{stream_id}` → `"1"` (directory for sweeper).
pub struct KvResumeBuffer {
    kv: Arc<dyn KvStore>,
    tuning: SweeperTuning,
}

/// Compact summary stored as the directory entry value so `list_all_streams`
/// can read all stream summaries in a single `scan_bucket` call.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct DirectorySummary {
    is_terminal: bool,
    next_id: u64,
    last_write_unix_ms: u64,
}

impl DirectorySummary {
    fn from_meta(meta: &BufferMeta) -> Self {
        Self {
            is_terminal: meta.terminal,
            next_id: meta.next_id,
            last_write_unix_ms: meta.last_write_unix_ms,
        }
    }
}

/// Per-stream metadata persisted as JSON under the `meta` key.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default)]
pub(crate) struct BufferMeta {
    /// Next id to assign. `next_id - 1` is the highest retained id.
    pub next_id: u64,
    /// Oldest retained id. `event/{oldest_id}` exists until ring
    /// eviction bumps it past the requested id.
    pub oldest_id: u64,
    /// Unix milliseconds of the most recent `record` call. Stored in
    /// milliseconds so the sweeper can use sub-second TTLs in tests.
    pub last_write_unix_ms: u64,
    /// `true` once `close` has been called. Used by the sweeper for
    /// terminal-grace eviction.
    pub terminal: bool,
}

impl KvResumeBuffer {
    /// Build from any `Arc<dyn KvStore>` with default sweeper tuning.
    pub fn new(kv: Arc<dyn KvStore>) -> Self {
        Self {
            kv,
            tuning: SweeperTuning::default(),
        }
    }

    /// Build with custom sweeper tuning. Use in tests with short TTLs.
    pub fn with_tuning(kv: Arc<dyn KvStore>, tuning: SweeperTuning) -> Self {
        Self { kv, tuning }
    }

    fn bucket(stream_id: &str) -> String {
        format!("{RESUME_BUCKET_PREFIX}{stream_id}")
    }

    fn event_key(id: u64) -> String {
        format!("{EVENT_KEY_PREFIX}{id}")
    }

    fn now_unix_ms() -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0)
    }

    pub(crate) async fn read_meta(&self, stream_id: &str) -> Result<BufferMeta, ResumeError> {
        let bucket = Self::bucket(stream_id);
        let entry = self
            .kv
            .get(&bucket, META_KEY)
            .await
            .map_err(|e| ResumeError::Backend(Box::new(e)))?;
        match entry {
            Some(entry) => serde_json::from_slice::<BufferMeta>(&entry.value)
                .map_err(|e| ResumeError::Backend(Box::new(e))),
            None => Err(ResumeError::NotFound(stream_id.to_string())),
        }
    }

    async fn write_meta(&self, stream_id: &str, meta: &BufferMeta) -> Result<(), ResumeError> {
        let bucket = Self::bucket(stream_id);
        let bytes = serde_json::to_vec(meta).map_err(|e| ResumeError::Backend(Box::new(e)))?;
        self.kv
            .put(&bucket, META_KEY, Bytes::from(bytes))
            .await
            .map_err(|e| ResumeError::Backend(Box::new(e)))?;
        Ok(())
    }
}

#[async_trait]
impl ResumeBuffer for KvResumeBuffer {
    #[tracing::instrument(
        skip_all,
        fields(
            db.system = "klieo-kv",
            db.namespace = "klieo-resume",
            db.operation = "record",
            klieo.resume.stream_id = %stream_id,
            klieo.resume.event_id = event_id,
        ),
        level = "debug",
        err,
    )]
    async fn record(
        &self,
        stream_id: &str,
        event_id: u64,
        payload: Bytes,
    ) -> Result<(), ResumeError> {
        let bucket = Self::bucket(stream_id);
        let mut meta = match self.read_meta(stream_id).await {
            Ok(m) => m,
            Err(ResumeError::NotFound(_)) => BufferMeta {
                next_id: event_id,
                oldest_id: event_id,
                last_write_unix_ms: Self::now_unix_ms(),
                terminal: false,
            },
            Err(other) => return Err(other),
        };

        // Write event payload before bumping next_id so a partial
        // failure here leaves meta untouched.
        self.kv
            .put(&bucket, &Self::event_key(event_id), payload)
            .await
            .map_err(|e| ResumeError::Backend(Box::new(e)))?;

        if event_id >= meta.next_id {
            meta.next_id = event_id + 1;
        }

        // Evict oldest slots while ring exceeds capacity.
        while meta.next_id.saturating_sub(meta.oldest_id) > RESUME_RING_CAP as u64 {
            let to_evict = meta.oldest_id;
            // Best-effort delete; benign if already absent.
            let _ = self.kv.delete(&bucket, &Self::event_key(to_evict)).await;
            meta.oldest_id = to_evict + 1;
        }

        meta.last_write_unix_ms = Self::now_unix_ms();
        self.write_meta(stream_id, &meta).await?;

        // Maintain the directory index so the sweeper can enumerate active
        // streams and list_all_streams can read summaries without N+1
        // read_meta calls. Best-effort; missing-index just disables
        // proactive eviction for this stream.
        let summary_bytes = match serde_json::to_vec(&DirectorySummary::from_meta(&meta)) {
            Ok(b) => Bytes::from(b),
            Err(e) => {
                tracing::error!(
                    target: "klieo.resume",
                    stream_id,
                    error = %e,
                    "directory summary serialization failed"
                );
                return Ok(());
            }
        };
        if let Err(e) = self
            .kv
            .put(RESUME_DIRECTORY_BUCKET, stream_id, summary_bytes)
            .await
        {
            tracing::warn!(
                target: "klieo.resume",
                stream_id,
                error = %e,
                "resume directory index write failed; sweeper may miss this stream"
            );
        }
        Ok(())
    }

    #[tracing::instrument(
        skip_all,
        fields(
            db.system = "klieo-kv",
            db.namespace = "klieo-resume",
            db.operation = "replay",
            klieo.resume.stream_id = %stream_id,
            klieo.resume.since_id = since_id,
        ),
        level = "debug",
        err,
    )]
    async fn replay(
        &self,
        stream_id: &str,
        since_id: u64,
    ) -> Result<Pin<Box<dyn Stream<Item = (u64, Bytes)> + Send>>, ResumeError> {
        let meta = self.read_meta(stream_id).await?;
        // Allow since_id + 1 == oldest_id (caller asks for events
        // starting at oldest_id; we still have them). Anything older
        // is expired.
        if since_id + 1 < meta.oldest_id {
            return Err(ResumeError::Expired { since_id });
        }
        let start = since_id + 1;
        let end = meta.next_id; // exclusive

        let bucket = Self::bucket(stream_id);
        // Eagerly collect retained events so we don't need the `futures` crate.
        let mut events: Vec<(u64, Bytes)> = Vec::new();
        for id in start..end {
            let key = Self::event_key(id);
            match self.kv.get(&bucket, &key).await {
                Ok(Some(entry)) => events.push((id, entry.value)),
                // Race: event evicted between meta read and get. Skip.
                Ok(None) => {}
                Err(e) => return Err(ResumeError::Backend(Box::new(e))),
            }
        }
        Ok(Box::pin(tokio_stream::iter(events)))
    }

    #[tracing::instrument(
        skip_all,
        fields(
            db.system = "klieo-kv",
            db.namespace = "klieo-resume",
            db.operation = "close",
            klieo.resume.stream_id = %stream_id,
        ),
        level = "debug",
        err,
    )]
    async fn close(&self, stream_id: &str) -> Result<(), ResumeError> {
        let mut meta = self.read_meta(stream_id).await?;
        meta.terminal = true;
        self.write_meta(stream_id, &meta).await?;
        // Best-effort update of directory summary so list_all_streams reflects
        // terminal state without a read_meta fallback.
        let summary_bytes = match serde_json::to_vec(&DirectorySummary::from_meta(&meta)) {
            Ok(b) => Bytes::from(b),
            Err(e) => {
                tracing::warn!(
                    target: "klieo.resume",
                    stream_id,
                    error = %e,
                    "close: directory summary serialization failed"
                );
                return Ok(());
            }
        };
        if let Err(e) = self
            .kv
            .put(RESUME_DIRECTORY_BUCKET, stream_id, summary_bytes)
            .await
        {
            tracing::warn!(
                target: "klieo.resume",
                stream_id,
                error = %e,
                "close: directory index update failed"
            );
        }
        Ok(())
    }

    #[tracing::instrument(
        skip_all,
        fields(
            db.system = "klieo-kv",
            db.namespace = "klieo-resume",
            db.operation = "is_terminal",
            klieo.resume.stream_id = %stream_id,
        ),
        level = "debug",
        err,
    )]
    async fn is_terminal(&self, stream_id: &str) -> Result<bool, ResumeError> {
        match self.read_meta(stream_id).await {
            Ok(meta) => Ok(meta.terminal),
            Err(ResumeError::NotFound(_)) => Ok(false),
            Err(e) => Err(e),
        }
    }
}

impl KvResumeBuffer {
    /// Enumerate all streams known to the resume-buffer directory.
    ///
    /// Returns entries for both active and terminal streams. Streams that
    /// have been evicted by the sweeper no longer appear.
    ///
    /// When the backing [`KvStore`] does not support key enumeration
    /// (`BusError::Unsupported`), returns an empty `Vec` — the same
    /// graceful degradation used by the sweeper.
    pub async fn list_all_streams(&self) -> Result<Vec<StreamEntry>, BusError> {
        let pairs = match self.kv.scan_bucket(RESUME_DIRECTORY_BUCKET).await {
            Ok(p) => p,
            Err(BusError::Unsupported(_)) => return Ok(vec![]),
            Err(e) => return Err(e),
        };

        let mut entries = Vec::with_capacity(pairs.len());
        for (stream_id, value) in pairs {
            if let Some(entry) = self.entry_from_directory(stream_id, &value).await {
                entries.push(entry);
            }
        }
        Ok(entries)
    }

    /// Enumerate streams one bounded page at a time, scoping the directory
    /// scan itself to `limit` keys via [`KvStore::keys_paginated`].
    ///
    /// Unlike [`Self::list_all_streams`], which loads the whole directory,
    /// this fetches at most `limit` directory keys per call — so an operator
    /// listing never materialises every stream at once. Pass `cursor = None`
    /// for the first page and the returned [`StreamPage::next`] thereafter;
    /// `next == None` means the scan is exhausted.
    ///
    /// When the backing [`KvStore`] cannot enumerate keys
    /// (`BusError::Unsupported`), returns an empty page — the same graceful
    /// degradation [`Self::list_all_streams`] applies to `scan_bucket`.
    pub async fn list_streams_paged(
        &self,
        cursor: Option<String>,
        limit: usize,
    ) -> Result<StreamPage, BusError> {
        let page = match self
            .kv
            .keys_paginated(RESUME_DIRECTORY_BUCKET, cursor, limit)
            .await
        {
            Ok(p) => p,
            Err(BusError::Unsupported(_)) => {
                return Ok(StreamPage {
                    entries: vec![],
                    next: None,
                })
            }
            Err(e) => return Err(e),
        };

        let mut entries = Vec::with_capacity(page.keys.len());
        for key in page.keys {
            // A key present in keys_paginated but absent on get is a raced
            // deletion between enumeration and fetch — skip it, don't error.
            let Some(entry) = self.kv.get(RESUME_DIRECTORY_BUCKET, &key).await? else {
                continue;
            };
            if let Some(built) = self.entry_from_directory(key, &entry.value).await {
                entries.push(built);
            }
        }
        Ok(StreamPage {
            entries,
            next: page.next,
        })
    }

    /// Prefers the compact `DirectorySummary` stored as the value; on a
    /// legacy (`b"1"`) or corrupt value it falls back to [`Self::read_meta`].
    /// Returns `None` when the stream has no meta (evicted between
    /// enumeration and this read) so callers drop it silently.
    async fn entry_from_directory(&self, stream_id: String, value: &[u8]) -> Option<StreamEntry> {
        match serde_json::from_slice::<DirectorySummary>(value) {
            Ok(summary) => Some(StreamEntry {
                stream_id,
                is_terminal: summary.is_terminal,
                next_id: summary.next_id,
                last_write_unix_ms: summary.last_write_unix_ms,
            }),
            Err(e) => {
                tracing::debug!(
                    target: "klieo.resume",
                    stream_id = %stream_id,
                    error = %e,
                    "directory entry not a summary, falling back to read_meta"
                );
                match self.read_meta(&stream_id).await {
                    Ok(meta) => Some(StreamEntry {
                        stream_id,
                        is_terminal: meta.terminal,
                        next_id: meta.next_id,
                        last_write_unix_ms: meta.last_write_unix_ms,
                    }),
                    Err(ResumeError::NotFound(_)) => None,
                    Err(other) => {
                        tracing::warn!(
                            target: "klieo.resume",
                            stream_id,
                            error = %other,
                            "fallback read_meta failed; skipping stream"
                        );
                        None
                    }
                }
            }
        }
    }

    /// Spawn a background task that periodically walks the resume
    /// directory bucket and evicts buckets idle past `tuning.idle_ttl`
    /// or terminal-past-grace (`tuning.terminal_grace`).
    ///
    /// The returned `JoinHandle` is owned by the caller; the task
    /// terminates when the supplied `cancel` token fires. For production,
    /// wire the token to the server's parent cancel so server shutdown
    /// cleanly stops the sweeper.
    ///
    /// Backends without [`KvStore::keys`] enumeration support return
    /// `BusError::Unsupported`; the sweeper logs once and becomes inert —
    /// record/replay still work, but proactive eviction is disabled.
    pub fn spawn_sweeper(
        self: Arc<Self>,
        cancel: tokio_util::sync::CancellationToken,
    ) -> tokio::task::JoinHandle<()> {
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(self.tuning.sweep_interval);
            // First tick fires immediately; skip it so the sweeper does
            // not run before any data has landed.
            tick.tick().await;
            loop {
                tokio::select! {
                    biased;
                    _ = cancel.cancelled() => break,
                    _ = tick.tick() => {
                        if let Err(e) = self.sweep_once().await {
                            tracing::warn!(
                                target: "klieo.resume.sweeper",
                                error = %e,
                                "resume buffer sweep failed; will retry next interval"
                            );
                        }
                    }
                }
            }
        })
    }

    async fn sweep_once(&self) -> Result<(), BusError> {
        let stream_ids = match self.kv.keys(RESUME_DIRECTORY_BUCKET).await {
            Ok(ids) => ids,
            Err(BusError::Unsupported(_)) => {
                tracing::debug!(
                    target: "klieo.resume.sweeper",
                    "KvStore backend does not support keys(); proactive eviction disabled"
                );
                return Ok(());
            }
            Err(e) => return Err(e),
        };

        let now_ms = Self::now_unix_ms();
        for stream_id in stream_ids {
            let meta = match self.read_meta(&stream_id).await {
                Ok(m) => m,
                // Genuinely stale entry: meta gone, directory entry leaked.
                Err(ResumeError::NotFound(_)) => {
                    if let Err(e) = self.kv.delete(RESUME_DIRECTORY_BUCKET, &stream_id).await {
                        tracing::warn!(
                            target: "klieo.resume.sweeper",
                            stream_id,
                            error = %e,
                            "stale directory-entry delete failed; will retry next interval"
                        );
                    }
                    continue;
                }
                // Backend error reading meta — DO NOT delete the directory
                // entry. A transient KV error would otherwise look like a
                // stale entry and orphan live event data with no visibility.
                Err(other) => {
                    tracing::warn!(
                        target: "klieo.resume.sweeper",
                        stream_id,
                        error = %other,
                        "read_meta failed during sweep; skipping this stream this interval"
                    );
                    continue;
                }
            };
            let age_ms = now_ms.saturating_sub(meta.last_write_unix_ms);
            let evict = if meta.terminal {
                age_ms >= self.tuning.terminal_grace.as_millis() as u64
            } else {
                age_ms >= self.tuning.idle_ttl.as_millis() as u64
            };
            if evict {
                self.evict_bucket(&stream_id).await?;
            }
        }
        Ok(())
    }

    async fn evict_bucket(&self, stream_id: &str) -> Result<(), BusError> {
        let bucket = Self::bucket(stream_id);
        let keys = match self.kv.keys(&bucket).await {
            Ok(k) => k,
            Err(BusError::Unsupported(_)) => return Ok(()),
            Err(e) => return Err(e),
        };
        let mut delete_failures: usize = 0;
        for k in keys {
            if let Err(e) = self.kv.delete(&bucket, &k).await {
                delete_failures += 1;
                tracing::warn!(
                    target: "klieo.resume.sweeper",
                    stream_id,
                    key = %k,
                    error = %e,
                    "evict delete failed; will retry next interval"
                );
            }
        }
        // Only purge the directory entry when every key delete succeeded.
        // Otherwise a partial eviction would orphan the surviving event/*
        // data with no sweeper visibility (data loss with no recovery
        // path next interval).
        if delete_failures == 0 {
            if let Err(e) = self.kv.delete(RESUME_DIRECTORY_BUCKET, stream_id).await {
                tracing::warn!(
                    target: "klieo.resume.sweeper",
                    stream_id,
                    error = %e,
                    "directory-entry purge failed; will retry next interval"
                );
            }
        } else {
            tracing::warn!(
                target: "klieo.resume.sweeper",
                stream_id,
                failures = delete_failures,
                "directory entry retained for retry; partial eviction"
            );
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn noop_record_is_ok() {
        let buf = NoopResumeBuffer;
        buf.record("s1", 1, Bytes::from_static(b"hi"))
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn noop_replay_returns_not_found() {
        let buf = NoopResumeBuffer;
        let result = buf.replay("s1", 0).await;
        match result {
            Err(ResumeError::NotFound(s)) => assert_eq!(s, "s1"),
            Err(other) => panic!("expected NotFound, got {other:?}"),
            Ok(_) => panic!("expected Err, got Ok"),
        }
    }

    #[tokio::test]
    async fn noop_close_is_ok() {
        let buf = NoopResumeBuffer;
        buf.close("s1").await.unwrap();
    }

    #[test]
    fn backend_error_preserves_source_chain() {
        use std::error::Error as _;
        let inner = std::io::Error::other("disk gone");
        let e = ResumeError::Backend(Box::new(inner));
        assert_eq!(e.to_string(), "buffer backend: disk gone");
        let src = e
            .source()
            .expect("Backend must preserve source for e.source()");
        assert_eq!(src.to_string(), "disk gone");
    }

    #[test]
    fn expired_renders_since_id() {
        let e = ResumeError::Expired { since_id: 42 };
        assert_eq!(
            e.to_string(),
            "resume window expired (oldest retained id > requested 42)"
        );
    }

    // ---- KvResumeBuffer tests ----

    fn in_mem_buffer() -> KvResumeBuffer {
        use crate::test_utils::fake_kv;
        KvResumeBuffer::new(fake_kv())
    }

    #[tokio::test]
    async fn kv_record_and_replay_in_order() {
        let buf = in_mem_buffer();
        for i in 1..=5u64 {
            buf.record("s1", i, Bytes::from(format!("e{i}")))
                .await
                .unwrap();
        }
        let stream = buf.replay("s1", 2).await.expect("replay");
        use tokio_stream::StreamExt as _;
        let mut stream = stream;
        let mut got = Vec::new();
        while let Some((id, bytes)) = stream.next().await {
            got.push((id, String::from_utf8(bytes.to_vec()).unwrap()));
        }
        assert_eq!(
            got,
            vec![(3, "e3".into()), (4, "e4".into()), (5, "e5".into())]
        );
    }

    #[tokio::test]
    async fn kv_ring_evicts_oldest_when_full() {
        let buf = in_mem_buffer();
        for i in 1..=300u64 {
            buf.record("s1", i, Bytes::from(format!("e{i}")))
                .await
                .unwrap();
        }
        // Cap is 256, so oldest_id == 300 - 256 + 1 == 45.
        // Replaying since 44 must succeed (next event is 45).
        let _ = buf.replay("s1", 44).await.expect("replay since 44");
        // Replaying since 5 (well before oldest=45) must return Expired.
        match buf.replay("s1", 5).await {
            Err(ResumeError::Expired { since_id }) => assert_eq!(since_id, 5),
            Err(other) => panic!("expected Expired, got Err({other:?})"),
            Ok(_) => panic!("expected Expired, got Ok(stream)"),
        }
    }

    #[tokio::test]
    async fn kv_replay_unknown_stream_returns_not_found() {
        let buf = in_mem_buffer();
        match buf.replay("missing", 0).await {
            Err(ResumeError::NotFound(s)) => assert_eq!(s, "missing"),
            Err(other) => panic!("expected NotFound, got Err({other:?})"),
            Ok(_) => panic!("expected NotFound, got Ok(stream)"),
        }
    }

    #[tokio::test]
    async fn kv_close_marks_terminal_in_meta() {
        let buf = in_mem_buffer();
        buf.record("s1", 1, Bytes::from_static(b"x")).await.unwrap();
        buf.close("s1").await.unwrap();
        let meta = buf.read_meta("s1").await.unwrap();
        assert!(meta.terminal);
    }

    #[tokio::test]
    async fn is_terminal_false_before_close() {
        let buf = in_mem_buffer();
        buf.record("s-1", 1, Bytes::from_static(b"hi"))
            .await
            .unwrap();
        assert!(!buf.is_terminal("s-1").await.unwrap());
    }

    #[tokio::test]
    async fn is_terminal_true_after_close() {
        let buf = in_mem_buffer();
        buf.record("s-1", 1, Bytes::from_static(b"hi"))
            .await
            .unwrap();
        buf.close("s-1").await.unwrap();
        assert!(buf.is_terminal("s-1").await.unwrap());
    }

    #[tokio::test]
    async fn is_terminal_false_on_unknown_id() {
        let buf = in_mem_buffer();
        assert!(!buf.is_terminal("never").await.unwrap());
    }

    // ---- Sweeper tests ----

    fn fast_sweeper_buffer() -> KvResumeBuffer {
        use crate::test_utils::fake_kv;
        KvResumeBuffer::with_tuning(
            fake_kv(),
            SweeperTuning {
                idle_ttl: Duration::from_millis(100),
                sweep_interval: Duration::from_millis(20),
                terminal_grace: Duration::from_millis(50),
            },
        )
    }

    #[tokio::test]
    async fn sweeper_evicts_idle_buckets_after_ttl() {
        let buf = Arc::new(fast_sweeper_buffer());
        buf.record("s1", 1, Bytes::from_static(b"x")).await.unwrap();
        assert!(buf.read_meta("s1").await.is_ok());

        let cancel = tokio_util::sync::CancellationToken::new();
        let sweep_handle = buf.clone().spawn_sweeper(cancel.clone());

        // idle_ttl=100ms + sweep_interval=20ms + headroom.
        tokio::time::sleep(Duration::from_millis(300)).await;

        match buf.read_meta("s1").await {
            Err(ResumeError::NotFound(_)) => {}
            Ok(_) => panic!("sweeper did not evict idle bucket"),
            Err(other) => panic!("unexpected error: {other:?}"),
        }

        cancel.cancel();
        let _ = tokio::time::timeout(Duration::from_secs(1), sweep_handle).await;
    }

    #[tokio::test]
    async fn sweeper_evicts_terminal_buckets_after_grace() {
        let buf = Arc::new(fast_sweeper_buffer());
        buf.record("s1", 1, Bytes::from_static(b"x")).await.unwrap();
        buf.close("s1").await.unwrap();

        let cancel = tokio_util::sync::CancellationToken::new();
        let sweep_handle = buf.clone().spawn_sweeper(cancel.clone());

        // terminal_grace=50ms + sweep_interval=20ms + headroom.
        tokio::time::sleep(Duration::from_millis(200)).await;

        assert!(matches!(
            buf.read_meta("s1").await,
            Err(ResumeError::NotFound(_))
        ));
        cancel.cancel();
        let _ = tokio::time::timeout(Duration::from_secs(1), sweep_handle).await;
    }

    #[tokio::test]
    async fn list_all_streams_empty_when_no_streams() {
        let buf = in_mem_buffer();
        let entries = buf.list_all_streams().await.unwrap();
        assert!(entries.is_empty());
    }

    #[tokio::test]
    async fn list_all_streams_returns_active_and_terminal() {
        use bytes::Bytes;
        let buf = in_mem_buffer();

        buf.record("stream-a", 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();
        buf.record("stream-a", 2, Bytes::from_static(b"evt"))
            .await
            .unwrap();
        buf.record("stream-b", 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();
        buf.close("stream-b").await.unwrap();

        let mut entries = buf.list_all_streams().await.unwrap();
        entries.sort_by(|a, b| a.stream_id.cmp(&b.stream_id));

        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].stream_id, "stream-a");
        assert!(!entries[0].is_terminal);
        assert_eq!(entries[0].next_id, 3);

        assert_eq!(entries[1].stream_id, "stream-b");
        assert!(entries[1].is_terminal);
        assert_eq!(entries[1].next_id, 2);
    }

    #[tokio::test]
    async fn list_all_streams_falls_back_to_read_meta_for_legacy_entries() {
        use crate::test_utils::FakeKvStore;
        use bytes::Bytes;

        let kv = Arc::new(FakeKvStore::new());
        let buf = KvResumeBuffer::new(kv.clone() as Arc<dyn crate::bus::KvStore>);

        buf.record("stream-new", 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();

        buf.record("stream-legacy", 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();
        kv.put(
            RESUME_DIRECTORY_BUCKET,
            "stream-legacy",
            Bytes::from_static(b"1"),
        )
        .await
        .unwrap();

        let mut entries = buf.list_all_streams().await.unwrap();
        entries.sort_by(|a, b| a.stream_id.cmp(&b.stream_id));

        assert_eq!(
            entries.len(),
            2,
            "both streams must appear regardless of directory entry format"
        );
        assert!(
            entries.iter().any(|e| e.stream_id == "stream-legacy"),
            "stream-legacy must be included via read_meta fallback"
        );
        assert!(
            entries.iter().any(|e| e.stream_id == "stream-new"),
            "stream-new must be included via normal directory summary path"
        );
    }

    #[tokio::test]
    async fn list_streams_paged_bounds_and_cursors() {
        use bytes::Bytes;
        const SEED_COUNT: usize = 5;
        const PAGE_LIMIT: usize = 2;

        let buf = in_mem_buffer();
        let mut seeded: Vec<String> = (0..SEED_COUNT).map(|i| format!("s{i}")).collect();
        for id in &seeded {
            buf.record(id, 1, Bytes::from_static(b"evt")).await.unwrap();
        }
        // keys_paginated returns keys ascending — assert against sorted order,
        // not insertion order, since the backend guarantees only the former.
        seeded.sort();

        let page1 = buf.list_streams_paged(None, PAGE_LIMIT).await.unwrap();
        assert_eq!(stream_ids(&page1), seeded[0..2]);
        let cursor1 = page1.next.expect("first page must have a next cursor");

        let page2 = buf
            .list_streams_paged(Some(cursor1), PAGE_LIMIT)
            .await
            .unwrap();
        assert_eq!(stream_ids(&page2), seeded[2..4]);
        let cursor2 = page2.next.expect("second page must have a next cursor");

        let page3 = buf
            .list_streams_paged(Some(cursor2), PAGE_LIMIT)
            .await
            .unwrap();
        assert_eq!(stream_ids(&page3), seeded[4..5]);
        assert!(
            page3.next.is_none(),
            "final page must report an exhausted cursor"
        );
    }

    #[tokio::test]
    async fn list_streams_paged_skips_legacy_or_corrupt() {
        use crate::test_utils::FakeKvStore;
        use bytes::Bytes;
        const PAGE_LIMIT: usize = 10;

        let kv = Arc::new(FakeKvStore::new());
        let buf = KvResumeBuffer::new(kv.clone() as Arc<dyn crate::bus::KvStore>);

        buf.record("stream-new", 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();
        // Legacy value (b"1") forces the read_meta fallback path; meta exists.
        buf.record("stream-legacy", 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();
        kv.put(
            RESUME_DIRECTORY_BUCKET,
            "stream-legacy",
            Bytes::from_static(b"1"),
        )
        .await
        .unwrap();
        // A directory key whose stream has no meta must be skipped, not errored.
        kv.put(
            RESUME_DIRECTORY_BUCKET,
            "stream-orphan",
            Bytes::from_static(b"1"),
        )
        .await
        .unwrap();

        let page = buf.list_streams_paged(None, PAGE_LIMIT).await.unwrap();
        let ids = stream_ids(&page);
        assert!(
            ids.contains(&"stream-new".to_string()),
            "summary entry must appear"
        );
        assert!(
            ids.contains(&"stream-legacy".to_string()),
            "legacy entry must appear via read_meta fallback"
        );
        assert!(
            !ids.contains(&"stream-orphan".to_string()),
            "orphan directory key without meta must be skipped"
        );
    }

    #[tokio::test]
    async fn list_streams_paged_empty_when_unsupported() {
        use crate::test_utils::NoopKv;
        // NoopKv does not override keys()/keys_paginated() → Unsupported.
        let buf = KvResumeBuffer::new(Arc::new(NoopKv) as Arc<dyn crate::bus::KvStore>);
        let page = buf.list_streams_paged(None, 10).await.unwrap();
        assert!(page.entries.is_empty());
        assert!(page.next.is_none());
    }

    /// Wraps a `KvStore` and returns `None` from `get` for exactly one
    /// directory key, while still listing that key from enumeration — the
    /// raced-deletion shape: a stream present in `keys_paginated` but gone by
    /// the time `list_streams_paged` fetches it.
    struct DropsOneKeyOnGet {
        inner: Arc<dyn KvStore>,
        raced_key: String,
    }

    #[async_trait]
    impl KvStore for DropsOneKeyOnGet {
        async fn get(
            &self,
            bucket: &str,
            key: &str,
        ) -> Result<Option<crate::bus::KvEntry>, BusError> {
            if bucket == RESUME_DIRECTORY_BUCKET && key == self.raced_key {
                return Ok(None);
            }
            self.inner.get(bucket, key).await
        }

        async fn put(
            &self,
            bucket: &str,
            key: &str,
            value: Bytes,
        ) -> Result<crate::bus::Revision, BusError> {
            self.inner.put(bucket, key, value).await
        }

        async fn cas(
            &self,
            bucket: &str,
            key: &str,
            value: Bytes,
            expected: Option<crate::bus::Revision>,
        ) -> Result<crate::bus::Revision, BusError> {
            self.inner.cas(bucket, key, value, expected).await
        }

        async fn delete(&self, bucket: &str, key: &str) -> Result<(), BusError> {
            self.inner.delete(bucket, key).await
        }

        async fn lease(
            &self,
            bucket: &str,
            key: &str,
            ttl: Duration,
        ) -> Result<crate::bus::Lease, BusError> {
            self.inner.lease(bucket, key, ttl).await
        }

        async fn keys(&self, bucket: &str) -> Result<Vec<String>, BusError> {
            self.inner.keys(bucket).await
        }
    }

    #[tokio::test]
    async fn list_streams_paged_skips_raced_deletion() {
        use crate::test_utils::FakeKvStore;
        const PAGE_LIMIT: usize = 10;
        const RACED: &str = "stream-raced";

        let backing = Arc::new(FakeKvStore::new());
        let seed = KvResumeBuffer::new(backing.clone() as Arc<dyn KvStore>);
        seed.record("stream-live", 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();
        seed.record(RACED, 1, Bytes::from_static(b"evt"))
            .await
            .unwrap();

        // The raced key is still enumerated, but its directory get returns None
        // — the deletion landed between enumeration and fetch.
        let raced_kv = Arc::new(DropsOneKeyOnGet {
            inner: backing as Arc<dyn KvStore>,
            raced_key: RACED.to_string(),
        });
        let buf = KvResumeBuffer::new(raced_kv as Arc<dyn KvStore>);

        let page = buf.list_streams_paged(None, PAGE_LIMIT).await.unwrap();
        let ids = stream_ids(&page);
        assert!(
            ids.contains(&"stream-live".to_string()),
            "surviving stream must still be returned"
        );
        assert!(
            !ids.contains(&RACED.to_string()),
            "raced-deleted stream must be silently skipped, not errored"
        );
    }

    fn stream_ids(page: &StreamPage) -> Vec<String> {
        page.entries.iter().map(|e| e.stream_id.clone()).collect()
    }
}