serve: --conversation-cache-spill-dir keeps evicted conversations on disk across restarts

CMakeLists.txt

@@ -114,6 +114,9 @@ if(STRATA_BUILD_CONVERSATION_TESTS)
   add_executable(conversation_file_test src/core/conversation_file_test.cpp src/core/conversation_file.cpp)
   target_include_directories(conversation_file_test PRIVATE include)
   add_test(NAME conversation_file_test COMMAND conversation_file_test)
+  add_executable(conversation_spill_test src/core/conversation_spill_test.cpp src/core/conversation_spill.cpp src/core/conversation_file.cpp)
+  target_include_directories(conversation_spill_test PRIVATE include)
+  add_test(NAME conversation_spill_test COMMAND conversation_spill_test)
   add_executable(session_save_reclaim_test src/core/session_save_reclaim_test.cpp src/core/conversation_file.cpp)
   target_include_directories(session_save_reclaim_test PRIVATE include)
   add_test(NAME session_save_reclaim_test COMMAND session_save_reclaim_test)
@@ -473,7 +476,7 @@ if(STRATA_ENABLE_CUDA OR STRATA_ENABLE_HIP)
     src/core/remote_expert_opt.cu
     src/core/expert_cache.cpp src/core/native_head.cpp src/core/native_dense.cpp src/core/verify.cpp src/core/mtp.cpp
     src/core/conversation_snapshot.cpp src/core/conversation_state.cpp src/core/conversation_memory.cpp src/core/peer_experts.cpp
-    src/core/arch_defaults.cpp src/core/vmm.cpp src/core/conversation_file.cpp)
+    src/core/arch_defaults.cpp src/core/vmm.cpp src/core/conversation_file.cpp src/core/conversation_spill.cpp)
   target_include_directories(strata_engine PUBLIC ${CMAKE_CURRENT_SOURCE_DIR}/include)
   target_link_libraries(strata_engine PUBLIC strata_core strata_kernels strata_kernels_cpu ${_strata_gpu_runtime_target})
   if(STRATA_BUILD_TESTS AND EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/tests/core/exchange_storage_test.cpp")

docs/DETAILS.md

@@ -1081,6 +1081,35 @@ sizes and K/V bytes reused during capture. `STRATA_SNAPSHOT_FULL_CAPTURE=1` disa
 retention for diagnostic comparisons. Parked snapshots are not
 persisted across restarts; the session files below are.
 
+**Disk tier for the parked conversations (opt-in).** Set `--conversation-cache-spill-dir DIR` to keep a parked
+conversation when RAM pressure or `--conversation-cache-slots` evicts it: the evicted conversation is written to DIR
+as an ordinary session file (the same format and model/configuration identity as the slot save/restore below), so a
+later request - or a restart - can read it back. The directory is scanned at the next start, and a request that
+matches a spilled conversation resumes from it even though it is no longer in RAM. The cache is still off unless
+`--conversation-cache-mib` is nonzero, and `--prompt-cache 0` or `--conversation-cache-slots 0` disables it.
+`--conversation-cache-disk-mib N` caps this model's spill files at 8192 MiB by default; `0` disables the disk tier.
+The index holds at most 256 conversations and evicts the oldest first. On a clean `QUIT` or stdin close, Strata
+captures the active conversation and spills the parked ones before it exits. Use a separate directory per model.
+
+Each spilled conversation is a session file (`.sess`) plus a small metadata sidecar (`.meta`) that holds only its
+token and image lists, so a new request finds the best disk match without reading the conversation's K/V. A file
+whose recorded model or configuration identity differs is refused, and a sidecar whose session file is missing, or
+whose own checksum fails, is removed when the directory is scanned. The identity is the same model fingerprint and
+configuration fingerprint the session files below are bound to, so a spilled conversation and a hand-saved session
+file are interchangeable.
+
+`--conversation-cache-similarity F` sets the least longest-common-prefix fraction of the new prompt a disk hit may
+offer (the fraction is `common_prefix_tokens / new_prompt_tokens`, and must be strictly greater than F);
+`--conversation-cache-n-min N` sets the least common-prefix token count. Both default to 0, which keeps every exact
+prefix the RAM cache would have used. Even when a candidate passes the threshold, Strata restores only a saved
+checkpoint whose token and image keys are an exact prefix of the new prompt; the last prompt token stays unread so
+the next verify window starts in the right position.
+
+A disk hit is read into host RAM before it is restored: it must fit the configured RAM cache budget and leave the
+`--conversation-cache-min-free-mib` physical-memory floor available, or the request reads the prompt normally. The
+engine logs spill, restore, stale-file and disk-budget events. The disk tier keeps the existing single-GPU parking
+limit: with `--layer-split` the disk tier stays off.
+
 **Session files (disk).** The conversation the engine holds can be saved to a file and restored later, also after a
 restart of the same engine version, so a long prompt is not read again. The server exposes the save and restore
 requests of llama-server's slot API, for its single slot 0, when started with `--slot-save-path DIR` (also

include/strata/core/conversation_cache.hpp

@@ -188,13 +188,104 @@ int64_t conversation_prefix(const ConversationCheckpoint& c, const std::vector<T
     return (int64_t) n;
 }
 
+// The same prefix test against only the first n ids of the checkpoint: a parked conversation's checkpoints are
+// prefixes of its live ids, so a shorter resume point is checked without copying the ids.
+template<class Token>
+int64_t conversation_prefix_at(const ConversationCheckpoint& c, const std::vector<Token>& prompt,
+                               const std::vector<ConversationImageKey>& images, size_t n) {
+    // The last prompt token always starts the next verify window.
+    if (n == 0 || n >= prompt.size() || n > c.ids.size() ||
+        !std::equal(c.ids.begin(), c.ids.begin() + n, prompt.begin())) return 0;
+    size_t a = 0, b = 0;
+    while ((a < c.imgs.size() && c.imgs[a].start < (int64_t) n) ||
+           (b < images.size() && images[b].start < (int64_t) n)) {
+        if (a == c.imgs.size() || c.imgs[a].start >= (int64_t) n ||
+            b == images.size() || images[b].start >= (int64_t) n || !(c.imgs[a] == images[b])) return 0;
+        ++a;
+        ++b;
+    }
+    return (int64_t) n;
+}
+
+// The longest common prefix of a cached conversation and the prompt, in tokens. Unlike conversation_prefix this
+// does not require the whole cached ids to match: it is the resume point a similarity hit would give.
+template<class Token>
+int64_t conversation_lcp(const ConversationCheckpoint& cached, const std::vector<Token>& prompt,
+                         const std::vector<ConversationImageKey>& images) {
+    size_t n = std::min(cached.ids.size(), prompt.size());
+    size_t i = 0;
+    while (i < n && cached.ids[i] == prompt[i]) ++i;
+    // Token ids alone do not identify image embeddings: stop before the first image boundary whose identity
+    // differs, so no state after it is reused.
+    size_t a = 0, b = 0;
+    while ((a < cached.imgs.size() && cached.imgs[a].start < (int64_t) i) ||
+           (b < images.size() && images[b].start < (int64_t) i)) {
+        if (a == cached.imgs.size() || cached.imgs[a].start >= (int64_t) i ||
+            b == images.size() || images[b].start >= (int64_t) i || !(cached.imgs[a] == images[b])) {
+            const int64_t ca = a < cached.imgs.size() ? cached.imgs[a].start : INT64_MAX;
+            const int64_t cb = b < images.size() ? images[b].start : INT64_MAX;
+            const int64_t boundary = std::min(ca, cb);
+            if (boundary < 0) return 0;
+            i = std::min(i, (size_t) boundary);
+            break;
+        }
+        ++a;
+        ++b;
+    }
+    return (int64_t) i;
+}
+
+struct ConversationMatch {
+    size_t index = 0;
+    int64_t tokens = 0;
+    bool live = false;
+};
+
+// The best resume a parked image offers for this prompt. similarity is the least LCP/new-prompt fraction accepted
+// (0 keeps every exact-prefix hit); n_min the least common-prefix tokens. A hit longer than the LCP is never offered.
+template<class Token>
+ConversationMatch conversation_saved_match(const SavedConversation& saved,
+                                           const std::vector<Token>& prompt,
+                                           const std::vector<ConversationImageKey>& images,
+                                           bool cvec, double similarity, int64_t n_min) {
+    ConversationMatch best;
+    if (saved.cvec != cvec || prompt.empty() || similarity < 0.0 || similarity >= 1.0) return best;
+    const int64_t lcp = conversation_lcp(saved.live, prompt, images);
+    if (lcp < n_min || (double) lcp / (double) prompt.size() <= similarity) return best;
+    auto consider = [&](const ConversationCheckpoint& checkpoint, bool live) {
+        const int64_t n = conversation_prefix(checkpoint, prompt, images);
+        if (n > 0 && n <= lcp && n > best.tokens) best = {0, n, live};
+    };
+    consider(saved.live, true);
+    for (const auto& checkpoint : saved.checkpoints) consider(checkpoint, false);
+    return best;
+}
+
+// The same match from a spill directory's metadata only (no K/V read): the live ids and images plus each
+// checkpoint's length, exactly what a parked image's checkpoints are prefixes of.
+template<class Token>
+ConversationMatch conversation_metadata_match(const ConversationCheckpoint& live,
+                                              const std::vector<size_t>& checkpoint_lengths,
+                                              const std::vector<Token>& prompt,
+                                              const std::vector<ConversationImageKey>& images,
+                                              bool cvec, bool cached_cvec,
+                                              double similarity, int64_t n_min) {
+    ConversationMatch best;
+    if (cached_cvec != cvec || prompt.empty() || similarity < 0.0 || similarity >= 1.0) return best;
+    const int64_t lcp = conversation_lcp(live, prompt, images);
+    if (lcp < n_min || (double) lcp / (double) prompt.size() <= similarity) return best;
+    const int64_t live_match = conversation_prefix(live, prompt, images);
+    if (live_match > 0 && live_match <= lcp) best = {0, live_match, true};
+    for (size_t length : checkpoint_lengths) {
+        const int64_t n = conversation_prefix_at(live, prompt, images, length);
+        if (n > 0 && n <= lcp && n > best.tokens) best = {0, n, false};
+    }
+    return best;
+}
+
 class ConversationCache {
 public:
-    struct Match {
-        size_t index = 0;
-        int64_t tokens = 0;
-        bool live = false;
-    };
+    using Match = ConversationMatch;
 
     ConversationCache(size_t budget, size_t slots, int64_t min_tokens = 0)
         : budget_(budget), slots_(slots), min_tokens_(min_tokens) {}
@@ -237,19 +328,15 @@ class ConversationCache {
     }
 
     template<class Token>
-    Match best(const std::vector<Token>& prompt, const std::vector<ConversationImageKey>& images, bool cvec) const {
+    Match best(const std::vector<Token>& prompt, const std::vector<ConversationImageKey>& images, bool cvec,
+               double similarity = 0.0, int64_t n_min = 0) const {
         Match best;
         // Ties prefer the most recently parked branch. The caller prefers its
         // already-active state when that offers the same prefix length.
         for (size_t i = entries_.size(); i-- > 0;) {
             const auto& e = entries_[i];
-            if (e.cvec != cvec) continue;
-            auto consider = [&](const ConversationCheckpoint& c, bool live) {
-                const int64_t n = conversation_prefix(c, prompt, images);
-                if (n > best.tokens) best = {i, n, live};
-            };
-            consider(e.live, true);
-            for (const auto& c : e.checkpoints) consider(c, false);
+            Match candidate = conversation_saved_match(e, prompt, images, cvec, similarity, n_min);
+            if (candidate.tokens > best.tokens) { candidate.index = i; best = candidate; }
         }
         return best;
     }
@@ -264,12 +351,19 @@ class ConversationCache {
     // Reserve before allocating a snapshot. held is an incoming image removed
     // with take() but still alive during the exchange; count it against RAM too.
     bool make_room(size_t incoming, size_t held = 0) {
+        return make_room(incoming, held, [](const SavedConversation&) {});
+    }
+
+    // The spill callback sees every conversation this call evicts from RAM, oldest first, before it is dropped, so
+    // a durable disk tier can keep it. The default does nothing (the RAM cache alone).
+    template<class Spill>
+    bool make_room(size_t incoming, size_t held, Spill&& spill) {
         if (!enabled() || held > budget_ || incoming > budget_ - held) return false;
         if (bytes() > budget_ - held - incoming) reuse_ = {};
         while (!entries_.empty() && (entries_.size() >= slots_ || bytes_ > budget_ - held - incoming)) {
             // the oldest entry that does not hold a pinned shared prefix leaves; with only pinned ones left the new
             // image does not fit (the caller skips parking it - the pinned prefix is what the queries come back to)
-            if (!evict_oldest()) return false;
+            if (!evict_oldest(spill)) return false;
         }
         return true;
     }
@@ -280,9 +374,15 @@ class ConversationCache {
     // own allocation, far above glibc's mmap threshold, so dropping an entry returns the whole footprint
     // to the kernel at once - the next admission check reads it back from /proc/meminfo.
     // False when none can go (empty, or only entries that hold a pinned shared prefix are left).
-    bool evict_oldest() {
+    bool evict_oldest() { return evict_oldest([](const SavedConversation&) {}); }
+
+    // The spill callback sees the conversation before it is dropped, so a durable disk tier keeps what the
+    // physical-RAM gate evicts and not only what make_room() pushes out.
+    template<class Spill>
+    bool evict_oldest(Spill&& spill) {
         auto victim = std::find_if(entries_.begin(), entries_.end(), [](const SavedConversation& e) { return !e.pinned(); });
         if (victim == entries_.end()) return false;
+        spill(*victim);
         bytes_ -= victim->bytes();
         entries_.erase(victim);
         ++evictions_;
@@ -336,6 +436,22 @@ class ConversationCache {
     }
     size_t superseded() const { return superseded_; }
 
+    // Spill every parked conversation (oldest first) and empty the cache: the shutdown path, so a restart finds
+    // them on disk. Returns how many were handed to the callback.
+    template<class Spill>
+    size_t spill_all(Spill&& spill) {
+        size_t count = 0;
+        while (!entries_.empty()) {
+            spill(entries_.front());
+            bytes_ -= entries_.front().bytes();
+            entries_.pop_front();
+            ++evictions_;
+            ++count;
+        }
+        reuse_ = {};
+        return count;
+    }
+
     bool put(SavedConversation&& image, size_t held = 0) {
         const size_t n = image.bytes();
         if (!enabled() || held > budget_ || n > budget_ - held) return false;   // make_room's refusal, first

include/strata/core/conversation_spill.hpp

@@ -0,0 +1,94 @@
+// A durable disk tier for --serve's parked conversations: when the RAM cache evicts a conversation, it is written
+// as an ordinary session file (conversation_file.hpp, the same format and model/config identity as the slot save/
+// restore API) and can be restored after a restart. A small sidecar beside each file holds only the token/image
+// metadata, so a new request finds the best disk match without reading the conversation's K/V.
+//
+// This is host code only (no CUDA): it calls the session-file writer and reader, which are host code too.
+#pragma once
+
+#include "strata/core/conversation_cache.hpp"
+#include "strata/core/conversation_file.hpp"
+
+#include <cstddef>
+#include <cstdint>
+#include <filesystem>
+#include <string>
+#include <vector>
+
+namespace strata::core {
+
+struct ConversationSpillMatch {
+    std::string path;        // the session file to read back
+    uint64_t file_bytes = 0;
+    int64_t tokens = 0;
+    bool live = false;
+    explicit operator bool() const { return !path.empty() && tokens > 0; }
+};
+
+class ConversationSpillCache {
+public:
+    // Opens (creating if needed) the spill directory and indexes the conversations already in it. Files whose
+    // sidecar names another model/config identity, or whose session file is missing or unreadable, are removed.
+    bool open(const std::filesystem::path& directory, SessionFileIdentity identity, uint64_t budget_bytes,
+              std::string& error);
+    bool enabled() const { return enabled_; }
+    size_t size() const { return entries_.size(); }
+    uint64_t bytes() const { return bytes_; }
+    size_t stale_files_wiped() const { return stale_files_wiped_; }
+    size_t disk_evictions() const { return disk_evictions_; }
+
+    // The best resume this directory offers for the prompt, from the sidecars only (no K/V read). similarity and
+    // n_min filter weak hits exactly as the RAM cache's best() does.
+    template<class Token>
+    ConversationSpillMatch best(const std::vector<Token>& prompt,
+                                const std::vector<ConversationImageKey>& images, bool cvec,
+                                double similarity, int64_t n_min) const {
+        ConversationSpillMatch best_match;
+        int64_t best_tokens = 0;
+        for (size_t i = entries_.size(); i-- > 0;) {
+            const Entry& entry = entries_[i];
+            const ConversationMatch match = conversation_metadata_match(entry.live_meta,
+                    entry.checkpoint_lengths, prompt, images, cvec, entry.cvec, similarity, n_min);
+            if (match.tokens > best_tokens)
+                best_tokens = match.tokens, best_match = {entry.session_path(), entry.file_bytes, match.tokens, match.live};
+        }
+        return best_match;
+    }
+
+    // Reads a spilled conversation back into image (its K/V included), with the engine's read limits and identity.
+    bool load(const std::string& path, SavedConversation& image, const SessionReadLimits& limits,
+              std::string& error) const;
+    // Writes a parked conversation (its K/V in RAM) as a session file plus its sidecar, and indexes it.
+    bool spill(const SavedConversation& image, std::string& error);
+    bool erase(const std::string& path, std::string& error);
+    // A conversation being restored is held out of eviction until it is put back in RAM or dropped.
+    void pin(const std::string& path);
+    void unpin(const std::string& path);
+    // Removes the disk copies of this conversation a turn back, the same rule the RAM cache applies before evicting.
+    size_t drop_superseded(const std::vector<int32_t>& ids, const std::vector<ConversationImageKey>& images,
+                           const std::vector<ConversationCheckpoint>& checkpoints, bool cvec);
+
+private:
+    struct Entry {
+        std::filesystem::path stem;          // "<dir>/strata-conv-<serial>" (both files share it)
+        uint64_t file_bytes = 0;
+        bool cvec = true;
+        ConversationCheckpoint live_meta;    // ids + imgs only; no running state
+        std::vector<size_t> checkpoint_lengths;
+        std::string session_path() const { return (stem.string() + ".sess"); }
+        std::string meta_path() const { return (stem.string() + ".meta"); }
+    };
+    void enforce_budget();
+    bool read_sidecar(const std::filesystem::path& meta, Entry& entry, bool& other_identity, std::string& error) const;
+    bool write_sidecar(const std::filesystem::path& meta, const Entry& entry, std::string& error) const;
+
+    bool enabled_ = false;
+    SessionFileIdentity identity_{};
+    uint64_t budget_ = 0, bytes_ = 0, serial_ = 0;
+    size_t stale_files_wiped_ = 0, disk_evictions_ = 0;
+    std::filesystem::path directory_;
+    std::vector<Entry> entries_;   // oldest spill first
+    std::string pinned_path_;
+};
+
+} // namespace strata::core