// CPU-only ownership and matching policy for --serve's parked conversations. // Token equality, image identity, and steering mode are all required for reuse. #pragma once #include "strata/core/conversation_buffer.hpp" #include #include #include #include #include #include #include namespace strata::core { struct ConversationImageKey { int64_t start = 0; uint64_t hash = 0; bool operator==(const ConversationImageKey&) const = default; }; struct ConversationCheckpoint { std::vector ids; std::vector imgs; std::vector gdn, ple, tails, dead, block_pos; uint64_t used = 0; // upstream root-pinned/LRU checkpoint retention // Ordinary layer-split checkpoints retain each device's running state. // Whole-session parking is currently single-GPU and rejects these parts. std::vector stage_parts; size_t bytes() const { size_t n = ids.capacity() * sizeof(int32_t) + imgs.capacity() * sizeof(ConversationImageKey) + gdn.capacity() + ple.capacity() + tails.capacity() + dead.capacity() + block_pos.capacity() + stage_parts.capacity() * sizeof(ConversationCheckpoint); for (const auto& part : stage_parts) n += part.bytes(); return n; } }; // Identity-layout K/V pages and completed indexer rows. For streamed layers the // source is the authoritative host pool, NOT the replaceable VRAM slots. struct ConversationKv { int format = 0; int64_t cells = 0, heads = 0, head_dim = 0, page_size = 0, pooled_rows = 0, idx_dim = 0; ConversationBuffer k, v, k_scale, v_scale, pooled; size_t bytes() const { return k.bytes() + v.bytes() + k_scale.bytes() + v_scale.bytes() + pooled.bytes(); } }; struct ConversationKvReuse { std::vector kv; // Original image extent for validation, and the earliest subsequent rewrite. int64_t captured_tokens = 0, unchanged_tokens = 0; size_t bytes() const { size_t n = kv.capacity() * sizeof(ConversationKv); for (const auto& layer : kv) n += layer.bytes(); return n; } }; struct SavedConversation { // Runtime compatibility only; NOT a model/weights identity or disk schema. std::array geometry{}; // The session's layer carve the image was captured from ([0, n_layers) on one GPU); restore requires the same. int64_t layer_lo = 0, layer_hi = 0; ConversationCheckpoint live; std::vector checkpoints; std::vector kv; // main layers followed by the draft layer bool cvec = true; size_t bytes() const { size_t n = live.bytes() + checkpoints.capacity() * sizeof(ConversationCheckpoint) + kv.capacity() * sizeof(ConversationKv); for (const auto& c : checkpoints) n += c.bytes(); for (const auto& k : kv) n += k.bytes(); return n; } }; template int64_t conversation_prefix(const ConversationCheckpoint& c, const std::vector& prompt, const std::vector& images) { const size_t n = c.ids.size(); // The last prompt token always starts the next verify window. if (n == 0 || n >= prompt.size() || !std::equal(c.ids.begin(), c.ids.end(), prompt.begin())) return 0; size_t j = 0; for (const auto& image : images) { if (image.start >= (int64_t) n) continue; if (j == c.imgs.size() || !(c.imgs[j++] == image)) return 0; } if (j != c.imgs.size()) return 0; return (int64_t) n; } class ConversationCache { public: struct Match { size_t index = 0; int64_t tokens = 0; bool live = false; }; ConversationCache(size_t budget, size_t slots) : budget_(budget), slots_(slots) {} bool enabled() const { return budget_ != 0 && slots_ != 0; } size_t bytes() const { return bytes_ + reuse_.bytes(); } size_t size() const { return entries_.size(); } size_t evictions() const { return evictions_; } // Retain only the restored K/V buffers, not duplicate running checkpoints. // This optimization never evicts a parked conversation to make itself fit. void retain(std::vector&& kv, int64_t tokens) { reuse_ = {}; ConversationKvReuse candidate{std::move(kv), tokens, tokens}; if (enabled() && candidate.bytes() <= budget_ - bytes_) reuse_ = std::move(candidate); } void limit_reuse(int64_t first_dirty) { reuse_.unchanged_tokens = std::min(reuse_.unchanged_tokens, first_dirty); if (reuse_.unchanged_tokens <= 0) reuse_ = {}; } ConversationKvReuse take_reuse() { return std::exchange(reuse_, {}); } size_t retained_bytes() const { return reuse_.bytes(); } bool can_fit(size_t incoming, size_t held = 0) const { return enabled() && held <= budget_ && incoming <= budget_ - held && entries_.size() < slots_ && bytes() <= budget_ - held - incoming; } template Match best(const std::vector& prompt, const std::vector& images, bool cvec) 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); } return best; } SavedConversation take(size_t index) { SavedConversation out = std::move(entries_.at(index)); bytes_ -= out.bytes(); entries_.erase(entries_.begin() + (std::ptrdiff_t) index); return out; } // 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) { 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)) { bytes_ -= entries_.front().bytes(); entries_.pop_front(); ++evictions_; } return true; } // #342: drop the parked entries an outgoing conversation (its live tokens and checkpoint chain) supersedes: // the same conversation a turn back, whose DEEPEST checkpoint the outgoing chain still holds, so all it adds // is the tail the client rewrote (the reply as it was generated, before the next request re-rendered it) and // checkpoints older than that one. A subagent's successive turns parked one such copy each, and make_room's // oldest-first eviction then pushed the parent conversation out after `slots` turns. An entry without // checkpoints, or whose deepest checkpoint the outgoing chain does not hold (another conversation that only // shares the system prompt's root with it), is kept. Returns how many were dropped. size_t drop_superseded(const std::vector& ids, const std::vector& images, const std::vector& checkpoints, bool cvec) { auto held = [&](const ConversationCheckpoint& c) { if (c.ids == ids && c.imgs == images) return true; for (const auto& k : checkpoints) if (k.ids == c.ids && k.imgs == c.imgs) return true; return false; }; size_t dropped = 0; for (size_t i = 0; i < entries_.size();) { const auto& e = entries_[i]; const ConversationCheckpoint* deepest = nullptr; for (const auto& c : e.checkpoints) if (!deepest || c.ids.size() > deepest->ids.size()) deepest = &c; if (e.cvec == cvec && deepest && !deepest->ids.empty() && held(*deepest)) { bytes_ -= e.bytes(); entries_.erase(entries_.begin() + (std::ptrdiff_t) i); ++dropped; continue; } ++i; } superseded_ += dropped; return dropped; } size_t superseded() const { return superseded_; } 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 drop_superseded(image.live.ids, image.live.imgs, image.checkpoints, image.cvec); if (!make_room(n, held)) return false; entries_.push_back(std::move(image)); bytes_ += n; return true; } private: size_t budget_ = 0, slots_ = 0, bytes_ = 0, evictions_ = 0, superseded_ = 0; std::deque entries_; // least recently active first ConversationKvReuse reuse_; }; } // namespace strata::core