Bound incremental snapshot fragmentation and preserve cache admission

Reserve accounted space for later appends without moving prior payloads. Prefer full capture when optional capacity would evict useful conversations, and report the actual additional RAM required. Exercise 4096 small appends and non-mutating admission checks under sanitizers.

include/strata/core/conversation_buffer.hpp

@@ -47,21 +47,36 @@ public:
     size_t allocation_peak(size_t n) const {
         size_t total = bytes();
         if (n <= size_) return total;
-        const size_t extra = n - size_;
-        if (!add(total, extra)) return SIZE_MAX;
+        const size_t spare = segments_.empty() ? 0 : segments_.back().capacity() - segments_.back().size();
+        const size_t extra = n - size_ - std::min(n - size_, spare);
+        const size_t full = extra / segment_bytes, tail = extra % segment_bytes;
+        if (!add(total, full * segment_bytes)) return SIZE_MAX;
+        if (tail && !add(total, segment_capacity(tail, full ? segment_bytes :
+                segments_.empty() ? 0 : segments_.back().capacity(), size_ != 0))) return SIZE_MAX;
         size_t count = segments_.size();
-        if (!add(count, extra / segment_bytes + (extra % segment_bytes != 0))) return SIZE_MAX;
+        if (!add(count, full + (tail != 0))) return SIZE_MAX;
         if (count > segments_.capacity() &&
             (count > SIZE_MAX / sizeof(Segment) || !add(total, count * sizeof(Segment)))) return SIZE_MAX;
         return total;
     }
     void resize(size_t n, uint8_t value = 0) {
         if (n > size_) {
-            const size_t extra = n - size_;
+            const bool growing = size_ != 0;
+            const size_t spare = segments_.empty() ? 0 : segments_.back().capacity() - segments_.back().size();
+            const size_t extend = std::min(n - size_, spare);
+            const size_t extra = n - size_ - extend;
             segments_.reserve(segments_.size() + extra / segment_bytes + (extra % segment_bytes != 0));
+            if (extend) {
+                segments_.back().resize(segments_.back().size() + extend, value);
+                size_ += extend;
+            }
             while (size_ < n) {
                 const size_t count = std::min(segment_bytes, n - size_);
-                segments_.emplace_back(count, value);
+                const size_t previous = segments_.empty() ? 0 : segments_.back().capacity();
+                Segment segment;
+                segment.reserve(segment_capacity(count, previous, growing));
+                segment.resize(count, value);
+                segments_.push_back(std::move(segment));
                 size_ += count;
             }
         } else {
@@ -105,6 +120,13 @@ public:
 
 private:
     using Segment = std::vector<uint8_t>;
+    // A fresh capture allocates exactly its payload. Later appends reserve
+    // geometrically growing segments, bounded by 16 MiB. Small chat turns then
+    // extend the final allocation instead of adding one transfer per turn.
+    static size_t segment_capacity(size_t count, size_t previous, bool growing) {
+        if (!growing) return count;
+        return std::max(count, std::min(segment_bytes, std::max<size_t>(65536, std::min(previous, segment_bytes/2)*2)));
+    }
     static bool add(size_t& n, size_t extra) {
         if (extra > SIZE_MAX - n) return false;
         n += extra;

include/strata/core/conversation_cache.hpp

@@ -118,6 +118,10 @@ public:
     }
     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<class Token>
     Match best(const std::vector<Token>& prompt, const std::vector<ConversationImageKey>& images, bool cvec) const {

src/core/conversation_cache_test.cpp

@@ -46,6 +46,17 @@ int main() {
         check(!bytes.read(tail.data(), bytes.size()-1, 2), "range check rejects a truncated payload");
         check(bytes.allocation_peak(SIZE_MAX) == SIZE_MAX, "allocation estimate rejects overflow");
     }
+    {
+        ConversationBuffer bytes;
+        for (size_t n=1;n<=4096;++n) {
+            const size_t peak = bytes.allocation_peak(n*17);
+            bytes.resize(n*17,7);
+            check(bytes.bytes() <= peak,"small append stays within the predicted allocation peak");
+        }
+        size_t segments = 0;
+        bytes.visit(0,bytes.size(),[&](const uint8_t*,size_t,size_t){++segments;return true;});
+        check(segments <= 4,"thousands of small turns do not create thousands of restore transfers");
+    }
     {
         ConversationKv layer;
         layer.k.resize(400);
@@ -62,6 +73,10 @@ int main() {
         check(reuse.unchanged_tokens == 9, "a later continuation cannot undo a rewind's dirty boundary");
         check(cache.bytes() == parked, "taking retained buffers releases their budget accounting");
         cache.retain(std::move(reuse.kv), 16);
+        check(!cache.can_fit(parked),"retained storage is counted when checking a capture without eviction");
+        check(cache.retained_bytes() == retained && cache.size() == 1 && cache.evictions() == 0,
+              "optional reuse admission does not evict or release anything");
+        check(!cache.can_fit(SIZE_MAX) && !cache.can_fit(1,SIZE_MAX),"non-mutating reservation rejects overflow");
         check(cache.make_room(parked), "reservation can discard optional active buffers");
         check(cache.retained_bytes() == 0 && cache.size() == 1 && cache.evictions() == 0,
               "pressure drops retained storage before evicting parked conversations");

src/program/generate.cpp

@@ -3278,8 +3278,9 @@ int main(int argc, char** argv) {
                 estimate = fresh_estimate;
                 err.clear();
             }
-            if (!conversations.make_room(estimate, held) && !reuse.kv.empty()) {
-                reuse = {}; // Optional retained storage must not prevent a full capture.
+            if (!reuse.kv.empty() && !conversations.can_fit(estimate, held)) {
+                // Optional growth capacity must not evict useful conversations.
+                reuse = {};
                 estimate = fresh_estimate;
             }
             if (!conversations.make_room(estimate, held)) {
@@ -3290,10 +3291,11 @@ int main(int argc, char** argv) {
             const auto t0 = Clock::now();
             try {
                 const uint64_t floor = (uint64_t) o.conversation_cache_min_free_mib * 1024 * 1024;
+                const size_t additional = estimate - reuse.bytes();
                 if (!strata::core::conversation_memory_admit(strata::core::conversation_available_memory(),
-                        estimate - reuse.bytes(), floor)) {
+                        additional, floor)) {
                     std::fprintf(stderr, "strata serve: conversation cache: skip parking (physical RAM admission; need %zu MiB plus %lld MiB floor, or telemetry unavailable)\n",
-                                 estimate >> 20, (long long) o.conversation_cache_min_free_mib);
+                                 additional >> 20, (long long) o.conversation_cache_min_free_mib);
                     return true;
                 }
                 strata::core::SavedConversation image;