perf: add opt-in Q8 expert buffer ownership rotation and validation

CMakeLists.txt

@@ -352,6 +352,13 @@ if(STRATA_ENABLE_CUDA OR STRATA_ENABLE_HIP)
     src/core/conversation_snapshot.cpp src/core/conversation_state.cpp src/core/conversation_memory.cpp src/core/peer_experts.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(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/tests/core/exchange_storage_test.cpp")
+    add_executable(exchange_storage_test tests/core/exchange_storage_test.cpp)
+    target_include_directories(exchange_storage_test PRIVATE include)
+    add_test(NAME exchange_storage_test COMMAND exchange_storage_test)
+    set_tests_properties(exchange_storage_test PROPERTIES TIMEOUT 60)
+  endif()
+
   if(EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/tests/core/file_expert_source_test.cpp")
     enable_testing()
     add_executable(file_expert_source_test tests/core/file_expert_source_test.cpp)

docs/DETAILS.md

@@ -12,6 +12,11 @@ New here? Start with the [README](../README.md); installing step by step is in [
 
 ---
 
+## Local Q8 buffer ownership experiment
+
+See [Q8 exchange rotation](Q8_EXCHANGE_ROTATION.md) for the opt-in storage change,
+correctness contract and test plan. Model speed results are pending.
+
 ## Speed (measured)
 
 RTX 5070 **12 GB**, Ryzen 5 7600 (6 cores), 64 GB DDR5-5200, Windows, engine 0.1.26 with the settings setup writes

docs/Q8_EXCHANGE_ROTATION.md

@@ -0,0 +1,72 @@
+# Q8 expert exchange buffer experiment
+
+Branch: `perf/q8-exchange-buffer-rotation`.
+Target: llm-60, NVIDIA RTX PRO 6000 Blackwell Workstation Edition **96 GB**,
+Ryzen 9 7950X, 128 GB installed RAM. Full Unsloth Q8_0, FP16 KV.
+
+**Status: implementation prepared; validation and throughput results pending.**
+There is no measured model speedup for this change yet.
+
+## The change
+
+The resident RAM mode keeps the experts absent from the GPU cache. Promoting
+expert A frees its RAM slot; evicting expert B first copies B from VRAM into a
+temporary host buffer. The original commit copies B again into A's former RAM
+slot. This experiment adopts B's temporary buffer as its resident storage and
+recycles A's former RAM slot as the next eviction buffer.
+
+Enable with `STRATA_EXCHANGE_ROTATE=1`. Unset or `0` retains the original copy
+path. The RAM and exchange allocations keep their original lifetime owners;
+only descriptors containing host and CUDA device addresses move. The model
+weights, resident expert count, cache replacement policy and quantization stay
+the same. The GPU-to-host eviction transfer is still required.
+
+This first implementation accepts **equal-size expert blocks with the entire
+resident complement pinned and mapped**. Partial pinning, pageable memory and
+mixed block sizes explicitly fall back to the copy path. The maximum exchange
+capacity must be reserved before rotation is initialized; later growth is
+rejected because the original exchange allocation may contain live experts.
+Existing serve/generate callers reserve their maximum once before inference.
+
+## Correctness contract
+
+- The D2H eviction finishes before the evicted expert is published for readers.
+- Existing CPU completion and H2D completion waits precede ownership commit.
+- Every resident expert and spare has exclusive ownership of one buffer.
+- CUDA aliases move with host pointers. No alias is reconstructed from a stale
+  offset into the original RAM allocation.
+- Staged overrides, file fallback, `blob`, `copy_blob`, prefetch, pin checks and
+  device address lookups resolve the same committed storage.
+- No transfer wait, residency publication order, stream lifetime or kernel
+  arithmetic changes. Invalid commit metadata fails closed.
+
+The counter `host memcpy bytes avoided` counts payload bytes. Avoiding an N-byte
+copy removes an N-byte CPU read and N-byte CPU write at the software level; it
+does not itself establish measured DRAM traffic or an end-to-end speedup.
+
+## Validation and measurement
+
+`exchange_storage_test` exercises 12,304 ownership transitions, odd sizes,
+unaligned host addresses, actual Q8-sized blocks, distinct device aliases,
+repeated reuse, invalid commits, guards and exact bytes. Run under ASan/UBSan.
+
+`file_expert_source_test --rotation-gpu` exercises the real source API and CUDA
+transfers with copy, rotation, and pageable fallback: staged reads, promotion,
+eviction, host copies, CUDA alias reads, disk fallback, growth rejection,
+close/reopen and byte equality. Run under Compute Sanitizer memcheck.
+
+`tools/run_q8_exchange_rotation_fleet.py` joins the private fleet GPU queue:
+
+1. Component tests, source API tests and GPU memory checks.
+2. Old binary versus new binary with rotation disabled, 8K input/256 output.
+3. Rotation off/on for serial, MTP, n-gram and combined modes, **64K input +
+   1,024 output**, 73,728 allocation, FP16 KV, fresh coding/editing requests.
+4. Fixed 16,400 GPU slots for serial/n-gram and 16,192 for MTP/combined; automatic
+   CPU/PCIe split, 96 adaptive swaps, completion waits enabled, ESP disabled.
+5. Exact token IDs, first divergence, resource telemetry, copy counters and
+   unprofiled decode speed. Reverse-order repeat for any exact-output candidate
+   with a first-run task improvement of at least 3%.
+
+A semantic mismatch remains a failed gate for that path, even if throughput
+increases. The storage fixtures and model comparison answer different questions;
+both are required before accepting the change. Public serving is unaffected.

include/strata/core/exchange_storage.hpp

@@ -0,0 +1,84 @@
+#pragma once
+
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+#include <string>
+#include <vector>
+
+namespace strata::core::detail {
+
+// Non-owning views: the complement and exchange allocations retain their original
+// owners until FileExpertSource::close(). A view moves; its bytes and CUDA alias do
+// not. This first path requires uniform, fully mapped/pinned expert slots.
+class ExchangeStorage {
+public:
+    struct View {
+        uint8_t* host = nullptr;
+        const uint8_t* device = nullptr;
+    };
+
+    bool initialize(const std::vector<uint64_t>& offsets, uint8_t* resident_host,
+                    const uint8_t* resident_device, uint64_t resident_bytes,
+                    uint8_t* exchange_host, const uint8_t* exchange_device,
+                    size_t count, size_t blob_bytes, std::string& error) {
+        error.clear();
+        if (active()) { error = "exchange storage already initialized"; return false; }
+        if (!resident_host || !resident_device || !exchange_host || !exchange_device ||
+            !blob_bytes || !count || !resident_bytes || resident_bytes % blob_bytes ||
+            count > std::numeric_limits<size_t>::max() / blob_bytes) {
+            error = "invalid uniform mapped exchange geometry"; return false;
+        }
+        std::vector<View> experts(offsets.size()), spares(count);
+        std::vector<bool> used((size_t)(resident_bytes / blob_bytes), false);
+        for (size_t i = 0; i < offsets.size(); ++i) {
+            const uint64_t at = offsets[i];
+            if (at == ~uint64_t{0}) continue;
+            if (at >= resident_bytes || at % blob_bytes || used[(size_t)(at / blob_bytes)]) {
+                error = "invalid or duplicate resident slot"; return false;
+            }
+            used[(size_t)(at / blob_bytes)] = true;
+            experts[i] = {resident_host + (size_t)at, resident_device + (size_t)at};
+        }
+        for (bool present : used) if (!present) {
+            error = "unassigned resident slot"; return false;
+        }
+        for (size_t q = 0; q < count; ++q)
+            spares[q] = {exchange_host + q * blob_bytes, exchange_device + q * blob_bytes};
+        experts_.swap(experts);
+        spares_.swap(spares);
+        blob_bytes_ = blob_bytes;
+        return true;
+    }
+
+    bool active() const { return blob_bytes_ != 0; }
+    View resident(size_t expert) const {
+        return expert < experts_.size() ? experts_[expert] : View{};
+    }
+    View spare(size_t q) const { return q < spares_.size() ? spares_[q] : View{}; }
+
+    // Only after every CPU reader and H2D copy of `in` has completed, and `out`
+    // has landed in spare(q). No allocation, memcpy, or CUDA operation here.
+    bool commit(size_t in, size_t out, size_t q, const uint8_t* staged, size_t bytes) {
+        if (in >= experts_.size() || out >= experts_.size() || q >= spares_.size() ||
+            !experts_[in].host || experts_[out].host || !staged ||
+            spares_[q].host != staged || bytes != blob_bytes_) return false;
+        experts_[out] = spares_[q];
+        spares_[q] = experts_[in];
+        experts_[in] = {};
+        ++exchanges_;
+        avoided_bytes_ += bytes;
+        return true;
+    }
+
+    uint64_t exchanges() const { return exchanges_; }
+    uint64_t avoided_bytes() const { return avoided_bytes_; } // memcpy payload, not read+write total
+    void clear() { experts_.clear(); spares_.clear(); blob_bytes_ = 0; exchanges_ = avoided_bytes_ = 0; }
+
+private:
+    std::vector<View> experts_, spares_;
+    size_t blob_bytes_ = 0;
+    uint64_t exchanges_ = 0, avoided_bytes_ = 0;
+};
+
+} // namespace strata::core::detail

include/strata/core/expert_source.hpp

@@ -22,6 +22,7 @@
 #pragma once
 
 #include "strata/core/expert_cache.hpp"
+#include "strata/core/exchange_storage.hpp"
 #include "strata/core/hit_hook.hpp"
 #include "strata/kernels/cpu/pool.hpp"
 
@@ -451,6 +452,7 @@ public:
     // buffer and calls `stage_exchange`: `out` is then read from that buffer, and `in` still from here (the CPU
     // computes both until the swap lands).  Once the slot copy has landed, `commit_exchanges` moves `out` into
     // `in`'s place, so the copy keeps holding exactly the experts the GPU does not - with no read of the file.
+    // STRATA_EXCHANGE_ROTATE=1 can instead transfer buffer ownership (uniform, fully pinned/mapped slots).
     /// Whether the compact copy holds `(layer, expert)`.
     bool has_resident(int64_t layer, int64_t expert) const;
     /// Host room for `n` evicted blobs (page-locked when possible).  Idempotent for the same or a smaller `n`.
@@ -462,6 +464,9 @@ public:
     /// After the GPU copies of every staged swap have landed.  Returns how many exchanges were applied.
     int64_t commit_exchanges();
     int64_t exchanges() const { return exchanges_; }
+    bool exchange_rotation() const { return exchange_storage_.active(); }
+    uint64_t rotated_exchanges() const { return exchange_storage_.exchanges(); }
+    uint64_t avoided_exchange_copy_bytes() const { return exchange_storage_.avoided_bytes(); }
     /// With the compact copy ready: blobs read from the mapped file since (what the plain mmap mode may read from
     /// the SSD).  0 in a steady resident mode; lend-region experts that did not fit the RAM count here.
     int64_t file_reads() const { return file_reads_.load(std::memory_order_relaxed); }
@@ -510,6 +515,7 @@ public:
     int64_t reads() const override { return reads_; }
 
 private:
+    const uint8_t* resident_blob(size_t index) const;
     const uint8_t* mapped_blob(int64_t layer, int64_t expert) const;
     /// The blob's bytes from the mapped file(s) - experts.bin, or the three GGUF role slices - into `dst`.
     bool copy_from_files(int64_t layer, int64_t expert, uint8_t* dst) const;
@@ -578,6 +584,7 @@ private:
     const uint8_t* complement_device_ = nullptr;
     uint64_t complement_bytes_ = 0;
     std::vector<uint64_t> complement_offsets_;
+    detail::ExchangeStorage exchange_storage_; // authoritative when active; original arenas still own memory
     bool complement_pinned_ = false;
     bool complement_partial_ = false;         ///< CS-T: only the first complement_pin_limit_ bytes are registered
     uint64_t complement_pin_limit_ = 0;

src/core/expert_source.cpp

@@ -478,6 +478,7 @@ void FileExpertSource::close() {
     xstage_blob_ = 0;
     override_.clear();
     staged_.clear();
+    exchange_storage_.clear();
     exchanges_ = 0;
     file_reads_.store(0);
     complement_arena_ = nullptr;
@@ -775,7 +776,7 @@ void FileExpertSource::prefetch(int64_t layer, const int64_t* experts, int64_t n
             const int64_t e = experts[i];
             if (e < 0 || e >= n_expert_) continue;
             const size_t index = (size_t) layer * (size_t) n_expert_ + (size_t) e;
-            if (complement_ready_ && index < complement_offsets_.size() && complement_offsets_[index] != kNoComplement)
+            if (complement_ready_ && resident_blob(index) != nullptr)
                 continue;                                     // in the RAM copy
             if (!override_.empty() && override_[index] != nullptr) continue;
             size_t v = 0;
@@ -1042,7 +1043,7 @@ void FileExpertSource::warm(int64_t layer, const int64_t* experts, int64_t n) {
         const int64_t e = experts[j];
         if (e < 0 || e >= n_expert_) continue;
         const size_t index = (size_t) layer * (size_t) n_expert_ + (size_t) e;
-        if (complement_ready_ && index < complement_offsets_.size() && complement_offsets_[index] != kNoComplement)
+        if (complement_ready_ && resident_blob(index) != nullptr)
             continue;                                                    // in the RAM copy
         if (warm_stamp_) warm_stamp_[index].store(stamp, std::memory_order_relaxed);
         warm_count_.fetch_add(1, std::memory_order_relaxed);
@@ -1158,7 +1159,7 @@ void FileExpertSource::begin_layer(int64_t layer, const int32_t* ids, int64_t k)
 bool FileExpertSource::transient(int64_t layer, int64_t expert) const {
     if (!staged() || layer < 0 || expert < 0 || layer >= n_layers_ || expert >= n_expert_) return false;
     const size_t index = (size_t) layer * (size_t) n_expert_ + (size_t) expert;
-    if (complement_ready_ && index < complement_offsets_.size() && complement_offsets_[index] != kNoComplement)
+    if (complement_ready_ && resident_blob(index) != nullptr)
         return false;
     return override_.empty() || override_[index] == nullptr;
 }
@@ -1170,7 +1171,7 @@ bool FileExpertSource::copy_blob(int64_t layer, int64_t expert, uint8_t* dst) {
     const uint64_t bytes = layer_blob_bytes_[(size_t) layer];
     if (complement_ready_) {
         const uint8_t* held =
-            detail::cache_complement_blob_or_fallback(index, complement_offsets_, complement_host_, nullptr);
+            resident_blob(index);
         if (held == nullptr && !override_.empty()) held = override_[index];
         if (held != nullptr) {
             std::memcpy(dst, held, (size_t) bytes);
@@ -1599,16 +1600,25 @@ bool FileExpertSource::pin_cache_complement(
     return true;
 }
 
+const uint8_t* FileExpertSource::resident_blob(size_t index) const {
+    if (exchange_storage_.active()) return exchange_storage_.resident(index).host;
+    return detail::cache_complement_blob_or_fallback(index, complement_offsets_, complement_host_, nullptr);
+}
+
 bool FileExpertSource::has_resident(int64_t layer, int64_t expert) const {
     if (!complement_ready_ || layer < 0 || expert < 0 || layer >= n_layers_ || expert >= n_expert_) return false;
     const size_t index = (size_t) layer * (size_t) n_expert_ + (size_t) expert;
-    return index < complement_offsets_.size() && complement_offsets_[index] != kNoComplement;
+    return resident_blob(index) != nullptr;
 }
 
 bool FileExpertSource::reserve_exchanges(int64_t n, std::string& err) {
     err.clear();
     if (n <= xstage_cap_) return true;
     if (!staged_.empty()) { err = "FileExpertSource: exchange buffers are in use"; return false; }
+    if (exchange_storage_.active()) {
+        err = "FileExpertSource: reserve maximum exchange capacity before enabling rotation";
+        return false; // the original exchange allocation can now contain live resident experts
+    }
     uint64_t blob = 0;
     for (const uint64_t b : layer_blob_bytes_) blob = std::max(blob, b);
     if (blob == 0 || n <= 0) { err = "FileExpertSource: no expert geometry for the exchange buffers"; return false; }
@@ -1618,9 +1628,18 @@ bool FileExpertSource::reserve_exchanges(int64_t n, std::string& err) {
         xstage_ = nullptr;
         xstage_cap_ = 0;
     }
+    if ((uint64_t)n > std::numeric_limits<size_t>::max() / blob) {
+        err = "FileExpertSource: exchange allocation size overflow"; return false;
+    }
+    const char* rotate_env = std::getenv("STRATA_EXCHANGE_ROTATE");
+    const bool requested = rotate_env && std::strcmp(rotate_env, "1") == 0;
+    const bool uniform = std::all_of(layer_blob_bytes_.begin(), layer_blob_bytes_.end(),
+                                    [blob](uint64_t b) { return b == blob; });
+    const bool eligible = requested && uniform && complement_pinned_ && !complement_partial_ &&
+                          complement_host_ && complement_device_ && complement_bytes_ > 0;
     const size_t total = (size_t) n * (size_t) blob;
     void* p = nullptr;
-    if (cudaHostAlloc(&p, total, cudaHostAllocDefault) == cudaSuccess && p != nullptr) {
+    if (cudaHostAlloc(&p, total, eligible ? cudaHostAllocMapped : cudaHostAllocDefault) == cudaSuccess && p != nullptr) {
         xstage_pinned_ = true;
     } else {
         (void) cudaGetLastError();
@@ -1631,16 +1650,36 @@ bool FileExpertSource::reserve_exchanges(int64_t n, std::string& err) {
     xstage_ = (uint8_t*) p;
     xstage_cap_ = n;
     xstage_blob_ = blob;
+    if (requested) {
+        std::string reason = "requires equal-size expert blocks and a fully mapped/pinned RAM complement";
+        void* device = nullptr;
+        if (eligible && xstage_pinned_ && cudaHostGetDevicePointer(&device, p, 0) == cudaSuccess && device) {
+            try {
+                if (exchange_storage_.initialize(complement_offsets_, (uint8_t*)complement_host_, complement_device_,
+                        complement_bytes_, xstage_, (const uint8_t*)device, (size_t)n, (size_t)blob, reason)) {
+                    std::fprintf(stderr, "FileExpertSource: exchange buffer rotation enabled: %lld buffers, %llu bytes each; no host commit memcpy\n",
+                                 (long long)n, (unsigned long long)blob);
+                }
+            } catch (const std::bad_alloc&) { reason = "buffer metadata allocation failed"; }
+        } else if (eligible) {
+            (void) cudaGetLastError();
+            reason = "exchange buffers could not be pinned and mapped";
+        }
+        if (!exchange_storage_.active())
+            std::fprintf(stderr, "FileExpertSource: exchange rotation unavailable; retaining copy path: %s\n", reason.c_str());
+    }
     return true;
 }
 
 uint8_t* FileExpertSource::exchange_buffer(int64_t q) const {
     if (xstage_ == nullptr || q < 0 || q >= xstage_cap_) return nullptr;
+    if (exchange_storage_.active()) return exchange_storage_.spare((size_t)q).host;
     return xstage_ + (size_t) q * (size_t) xstage_blob_;
 }
 
 bool FileExpertSource::stage_exchange(int64_t layer, int64_t in, int64_t out, int64_t q) {
-    if (!has_resident(layer, in) || has_resident(layer, out) || exchange_buffer(q) == nullptr) return false;
+    if (out < 0 || out >= n_expert_ || !has_resident(layer, in) || has_resident(layer, out) ||
+        exchange_buffer(q) == nullptr) return false;
     const size_t i_in = (size_t) layer * (size_t) n_expert_ + (size_t) in;
     const size_t i_out = (size_t) layer * (size_t) n_expert_ + (size_t) out;
     if (override_.empty()) override_.assign((size_t) blobs_, nullptr);
@@ -1656,6 +1695,16 @@ int64_t FileExpertSource::commit_exchanges() {
     int64_t n = 0;
     for (const Exchange& x : staged_) {
         const uint8_t* src = override_[x.out];
+        if (exchange_storage_.active()) {
+            if (!exchange_storage_.commit(x.in, x.out, (size_t)x.q, src, (size_t)x.bytes)) {
+                // Admitting the GPU swap after a failed RAM ownership update would lose an expert.
+                std::fprintf(stderr, "FileExpertSource: invalid exchange rotation commit; refusing corrupt residency\n");
+                std::abort();
+            }
+            ++n;
+            override_[x.out] = nullptr;
+            continue;
+        }
         const uint64_t at = complement_offsets_[x.in];
         if (src != nullptr && at != kNoComplement && at <= complement_bytes_ && x.bytes <= complement_bytes_ - at &&
             complement_host_ != nullptr) {
@@ -1675,7 +1724,7 @@ const uint8_t* FileExpertSource::blob(int64_t layer, int64_t expert) {
     const uint8_t* result = nullptr;
     bool from_files = true;
     if (complement_ready_) {
-        result = detail::cache_complement_blob_or_fallback(index, complement_offsets_, complement_host_, nullptr);
+        result = resident_blob(index);
         if (result != nullptr) {
             ram_reads_.fetch_add(1, std::memory_order_relaxed);
             from_files = false;
@@ -1702,6 +1751,7 @@ bool FileExpertSource::pinned(int64_t layer, int64_t expert) const {
     if (!complement_ready_ || !complement_pinned_ || complement_host_ == nullptr || layer < 0 || expert < 0 ||
         layer >= n_layers_ || expert >= n_expert_) return false;
     const size_t index = (size_t) layer * (size_t) n_expert_ + (size_t) expert;
+    if (exchange_storage_.active()) return exchange_storage_.resident(index).host != nullptr;
     if (index >= complement_offsets_.size() || complement_offsets_[index] == kNoComplement) return false;
     // a partial pin (CS-T): only the registered prefix
     return !complement_partial_ ||
@@ -1711,6 +1761,7 @@ bool FileExpertSource::pinned(int64_t layer, int64_t expert) const {
 const uint8_t* FileExpertSource::device_alias(int64_t layer, int64_t expert) const {
     if (!pinned(layer, expert) || complement_device_ == nullptr) return nullptr;
     const size_t index = (size_t) layer * (size_t) n_expert_ + (size_t) expert;
+    if (exchange_storage_.active()) return exchange_storage_.resident(index).device;
     return complement_device_ + (size_t) complement_offsets_[index];
 }
 

src/program/generate.cpp

@@ -6022,6 +6022,10 @@ int main(int argc, char** argv) {
                                      "VRAM tier, %lld blob reads from the file\n",
                              (double) src.resident_bytes() / 1073741824.0, (long long) src.exchanges(),
                              (long long) src.file_reads());
+            if (src.exchange_rotation())
+                std::fprintf(stderr, "strata serve: exchange rotation: %llu blocks, %llu host memcpy bytes avoided (cumulative payload)\n",
+                             (unsigned long long)src.rotated_exchanges(),
+                             (unsigned long long)src.avoided_exchange_copy_bytes());
             // CS-T: the tiers, cumulative - GPU cache hits (the decode lookups above), RAM copy, files (SSD / OS cache)
             if (srcp == &src)
                 std::fprintf(stderr, "strata serve: expert tiers: GPU %lld hits this request; since the start RAM %lld blobs, files %lld blobs "

tests/core/exchange_storage_test.cpp

@@ -0,0 +1,107 @@
+#include "strata/core/exchange_storage.hpp"
+
+#include <algorithm>
+#include <cstring>
+#include <iostream>
+#include <random>
+#include <set>
+#include <stdexcept>
+#include <vector>
+
+using strata::core::detail::ExchangeStorage;
+constexpr uint64_t absent = ~uint64_t{0};
+void require(bool ok, const char* why) { if (!ok) throw std::runtime_error(why); }
+
+void exercise(size_t bytes, size_t rounds) {
+    constexpr size_t n = 24, resident = 16, spare_count = 4, guard = 19;
+    std::vector<uint8_t> arena(guard + resident * bytes + guard, 0xcd);
+    std::vector<uint8_t> spare(guard + spare_count * bytes + guard, 0xef);
+    // Distinct alias arrays catch accidentally retaining the old CUDA address.
+    std::vector<uint8_t> aliases(resident * bytes), spare_aliases(spare_count * bytes);
+    std::vector<std::vector<uint8_t>> reference(n, std::vector<uint8_t>(bytes));
+    std::mt19937 random(601);
+    for (auto& row : reference) for (auto& v : row) v = (uint8_t)random();
+    std::vector<uint64_t> offsets(n, absent);
+    for (size_t i = 0; i < resident; ++i) {
+        offsets[i] = i * bytes;
+        std::memcpy(arena.data() + guard + offsets[i], reference[i].data(), bytes);
+    }
+    ExchangeStorage storage;
+    std::string error;
+    require(storage.initialize(offsets, arena.data() + guard, aliases.data(), resident * bytes,
+        spare.data() + guard, spare_aliases.data(), spare_count, bytes, error), error.c_str());
+    std::set<uint8_t*> original;
+    for (size_t i = 0; i < n; ++i) if (auto p = storage.resident(i).host) original.insert(p);
+    for (size_t q = 0; q < spare_count; ++q) original.insert(storage.spare(q).host);
+    uint64_t exchanges = 0;
+    for (size_t round = 0; round < rounds; ++round) {
+        std::vector<size_t> ins, outs;
+        for (size_t i = 0; i < n; ++i) (storage.resident(i).host ? ins : outs).push_back(i);
+        std::shuffle(ins.begin(), ins.end(), random);
+        std::shuffle(outs.begin(), outs.end(), random);
+        ExchangeStorage::View before[spare_count], staged[spare_count];
+        for (size_t q = 0; q < spare_count; ++q) {
+            before[q] = storage.resident(ins[q]); staged[q] = storage.spare(q);
+            // Simulated completed D2H eviction. The old RAM input stays readable
+            // until simulated H2D and all readers finish, before commit.
+            std::memcpy(staged[q].host, reference[outs[q]].data(), bytes);
+            require(!std::memcmp(before[q].host, reference[ins[q]].data(), bytes), "incoming expert overwritten early");
+        }
+        for (size_t q = 0; q < spare_count; ++q) {
+            require(!storage.commit(ins[q], ins[q], q, staged[q].host, bytes), "self swap accepted");
+            require(!storage.commit(ins[q], outs[q], q, staged[q].host, bytes + 1), "wrong size accepted");
+            require(!storage.commit(ins[q], outs[q], q, nullptr, bytes), "missing staged buffer accepted");
+            require(!storage.commit(ins[q], outs[q], spare_count, staged[q].host, bytes), "bad spare accepted");
+            require(!storage.commit(n, outs[q], q, staged[q].host, bytes), "bad expert accepted");
+            require(storage.commit(ins[q], outs[q], q, staged[q].host, bytes), "valid swap rejected");
+            require(!storage.commit(ins[q], outs[q], q, staged[q].host, bytes), "duplicate commit accepted");
+            require(storage.resident(outs[q]).host == staged[q].host &&
+                    storage.resident(outs[q]).device == staged[q].device, "evicted buffer was not adopted with its alias");
+            require(storage.spare(q).host == before[q].host && storage.spare(q).device == before[q].device,
+                    "old RAM input was not recycled with its alias");
+            // Even the old incoming bytes have not changed: commit copies metadata only.
+            require(!std::memcmp(storage.spare(q).host, reference[ins[q]].data(), bytes), "commit changed spare bytes");
+            ++exchanges;
+        }
+        std::set<uint8_t*> live;
+        for (size_t i = 0; i < n; ++i) if (auto p = storage.resident(i).host) {
+            require(live.insert(p).second, "two experts share storage");
+            require(!std::memcmp(p, reference[i].data(), bytes), "resident bytes corrupted");
+        }
+        for (size_t q = 0; q < spare_count; ++q)
+            require(live.insert(storage.spare(q).host).second, "spare aliases live expert");
+        require(live == original, "storage leaked or duplicated");
+    }
+    require(storage.exchanges() == exchanges && storage.avoided_bytes() == exchanges * bytes, "wrong accounting");
+    require(std::all_of(arena.begin(), arena.begin() + guard, [](auto x) { return x == 0xcd; }) &&
+            std::all_of(arena.end() - guard, arena.end(), [](auto x) { return x == 0xcd; }) &&
+            std::all_of(spare.begin(), spare.begin() + guard, [](auto x) { return x == 0xef; }) &&
+            std::all_of(spare.end() - guard, spare.end(), [](auto x) { return x == 0xef; }), "guard overwritten");
+    require(!storage.initialize(offsets, arena.data() + guard, aliases.data(), resident * bytes,
+        spare.data() + guard, spare_aliases.data(), spare_count, bytes, error), "live storage reinitialized");
+    storage.clear();
+    require(!storage.active() && !storage.resident(0).host && !storage.spare(0).host &&
+            !storage.exchanges() && !storage.avoided_bytes(), "clear retained state");
+}
+
+void invalid_geometry() {
+    uint8_t h[64]{}, d[64]{}, x[64]{}, xd[64]{};
+    ExchangeStorage s; std::string e;
+    require(!s.initialize({0, 3}, h, d, 8, x, xd, 2, 4, e), "unaligned slot accepted");
+    require(!s.initialize({0, 0}, h, d, 8, x, xd, 2, 4, e), "duplicate offset accepted");
+    require(!s.initialize({0, 8}, h, d, 8, x, xd, 2, 4, e), "out of bounds accepted");
+    require(!s.initialize({0, absent}, h, d, 8, x, xd, 2, 4, e), "lost slot accepted");
+    require(!s.initialize({0, 4}, h, nullptr, 8, x, xd, 2, 4, e), "unmapped resident accepted");
+    require(!s.initialize({0, 4}, h, d, 8, x, nullptr, 2, 4, e), "unmapped spare accepted");
+    require(!s.initialize({0, 4}, h, d, 8, x, xd, 2, 0, e), "zero blob accepted");
+    require(!s.active(), "failed initialization left active state");
+}
+
+int main() {
+    try {
+        invalid_geometry();
+        for (size_t bytes : {1u, 33u, 4097u}) exercise(bytes, 1024);
+        exercise(5222400, 4); // actual Q8 expert size
+        std::cout << "exchange_storage_test: PASS (byte/alias/ownership checks, 12304 exchanges)\n";
+    } catch (const std::exception& e) { std::cerr << e.what() << '\n'; return 1; }
+}

tests/core/file_expert_source_test.cpp

@@ -7,6 +7,9 @@
 #include "ggml.h"
 #endif
 
+#include <cuda_runtime.h>
+#include <cstdlib>
+#include <cstring>
 #include <algorithm>
 #include <chrono>
 #include <cstddef>
@@ -288,9 +291,97 @@ void test_cgroup_memory_budget() {
             "missing memory.stat counters did not fail closed");
 }
 
+
+void test_rotating_source(bool rotate, bool pin) {
+    using namespace strata::core;
+    using namespace strata::kernels::cpu;
+#if defined(_WIN32)
+    _putenv_s("STRATA_EXCHANGE_ROTATE", rotate ? "1" : "0");
+#else
+    setenv("STRATA_EXCHANGE_ROTATE", rotate ? "1" : "0", 1);
+#endif
+    TempDirectory dir;
+    std::string err;
+    require(expert_layout_load(dir.path.string(), 1, 5, err), err);
+    const size_t bytes = (size_t)BLOB;
+    std::vector<std::vector<uint8_t>> truth(5, std::vector<uint8_t>(bytes));
+    {
+        std::ofstream out(dir.path / "experts.bin", std::ios::binary);
+        for (size_t e = 0; e < truth.size(); ++e) {
+            for (size_t i = 0; i < bytes; ++i) truth[e][i] = (uint8_t)((i * 131u + e * 37u) ^ (i >> 8));
+            out.write((const char*)truth[e].data(), (std::streamsize)bytes);
+        }
+        require((bool)out, "fixture write failed");
+    }
+    FileExpertSource src;
+    ExpertCache cache;
+    require(src.open(dir.path.string(), 1, 5, err), err);
+    require(cache.open(2, 1, 5, BLOB, err), err);
+    for (int e : {2, 3}) {
+        const int slot = cache.admit(0, e);
+        require(slot >= 0 && cache.fill_slot_blocking(slot, truth[e].data(), err), err);
+    }
+    const std::vector<std::pair<int32_t, int32_t>> rank{{0,0},{0,1},{0,2},{0,3},{0,4}};
+    // Keep expert 4 on the file tier to check that the fallback survives rotation.
+    require(src.pin_cache_complement(cache, err, pin, {}, -1, 0, 2 * bytes, &rank), err);
+    require(src.reserve_exchanges(2, err), err);
+    require(src.exchange_rotation() == (rotate && pin), "rotation activation/fallback wrong");
+    require(src.reserve_exchanges(1, err), "smaller capacity rejected");
+    if (rotate && pin) require(!src.reserve_exchanges(3, err), "live arena growth accepted");
+    int incoming[2] = {0, 1}, outgoing[2] = {2, 3};
+    std::vector<uint8_t> actual(bytes);
+    double commit_ms = 0;
+    for (int round = 0; round < 32; ++round) {
+        const uint8_t* prior[2]{};
+        uint8_t* eviction[2]{};
+        for (int q = 0; q < 2; ++q) {
+            prior[q] = src.blob(0, incoming[q]); eviction[q] = src.exchange_buffer(q);
+            require(prior[q] && !std::memcmp(prior[q], truth[incoming[q]].data(), bytes), "wrong incoming bytes");
+            require(cudaMemcpy(eviction[q], cache.device_slot(q), bytes, cudaMemcpyDeviceToHost) == cudaSuccess,
+                    "D2H eviction failed");
+            require(src.stage_exchange(0, incoming[q], outgoing[q], q), "stage rejected");
+            require(!std::memcmp(src.blob(0, outgoing[q]), truth[outgoing[q]].data(), bytes), "staged override wrong");
+            require(cache.fill_slot_blocking(q, prior[q], err), err); // H2D completed before commit
+        }
+        require(!src.stage_exchange(0, incoming[0], outgoing[0], 0), "duplicate stage accepted");
+        require(!src.stage_exchange(0, incoming[0], 5, 0), "invalid expert accepted");
+        const auto t0 = std::chrono::steady_clock::now();
+        require(src.commit_exchanges() == 2, "commit count wrong");
+        commit_ms += std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t0).count();
+        require(src.commit_exchanges() == 0, "empty commit not empty");
+        for (int q = 0; q < 2; ++q) {
+            require(!src.has_resident(0, incoming[q]) && src.has_resident(0, outgoing[q]), "residency wrong");
+            const uint8_t* held = src.blob(0, outgoing[q]);
+            require(!std::memcmp(held, truth[outgoing[q]].data(), bytes), "evicted bytes corrupted");
+            require(held == ((rotate && pin) ? eviction[q] : prior[q]), "wrong storage selected");
+            if (rotate && pin) require(src.exchange_buffer(q) == prior[q], "old input not recycled");
+            require(src.copy_blob(0, outgoing[q], actual.data()) && actual == truth[outgoing[q]], "copy_blob mismatch");
+            require(!src.transient(0, outgoing[q]), "resident became transient");
+            if (pin) {
+                require(src.pinned(0, outgoing[q]) && src.device_alias(0, outgoing[q]), "mapping lost");
+                require(cudaMemcpy(actual.data(), src.device_alias(0, outgoing[q]), bytes, cudaMemcpyDeviceToHost) == cudaSuccess,
+                        "device alias read failed");
+                require(actual == truth[outgoing[q]], "CUDA alias points at stale expert");
+            }
+            require(cache.verify_slot(q, truth[incoming[q]].data(), err), err);
+            std::swap(incoming[q], outgoing[q]);
+        }
+        require(!src.has_resident(0, 4) && src.copy_blob(0, 4, actual.data()) && actual == truth[4], "file fallback changed");
+    }
+    require(src.exchanges() == 64, "exchange accounting wrong");
+    require(src.rotated_exchanges() == ((rotate && pin) ? 64u : 0u), "rotation accounting wrong");
+    require(src.avoided_exchange_copy_bytes() == ((rotate && pin) ? 64 * bytes : 0u), "avoided bytes wrong");
+    std::cout << "rotation integration: rotate=" << rotate << " pinned=" << pin
+              << " exchanges=64 commit_ms=" << commit_ms << " exact_bytes=PASS\n";
+    src.close();
+    require(!src.exchange_rotation() && !src.exchange_buffer(0), "close retained storage");
+    require(src.open(dir.path.string(), 1, 5, err), err);
+    require(!src.has_resident(0, 0) && !src.exchange_rotation(), "reopen retained residency");
+}
+
 }  // namespace
 
-int main() {
+int main(int argc, char** argv) {
     try {
         test_complement_plan();
         test_resident_lend_region();
@@ -300,6 +391,11 @@ int main() {
 #if defined(STRATA_NATIVE_EXPERTS)
         test_native_variable_layout();
 #endif
+        if (argc == 2 && std::string(argv[1]) == "--rotation-gpu") {
+            test_rotating_source(false, true);
+            test_rotating_source(true, true);
+            test_rotating_source(true, false);
+        }
         std::cout << "file_expert_source_test: PASS\n";
         return 0;
     } catch (const std::exception& error) {

tools/run_q8_exchange_rotation_fleet.py

@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+"""Q8/FP16 buffer ownership experiment, on the private llm-60 GPU queue."""
+from pathlib import Path
+import datetime
+import fcntl
+import hashlib
+import json
+import os
+import re
+import subprocess
+import sys
+import time
+
+H=Path.home(); R=Path(__file__).resolve().parents[1]
+PY=H/'src/Strata/.venv/bin/python'
+D=Path(os.environ.get('FLEET_EXCHANGE_OUTPUT',H/'fleet-downloads/rtxpro-q8-exchange-rotation-20261003'))
+PRIOR=Path(os.environ.get('FLEET_EXCHANGE_AFTER',H/'fleet-downloads/rtxpro-q8-long-context-retrieval-20261003/status.json'))
+CONTROL=H/'src/strata-q8-placement-080891d4'
+MODES={'serial':('off',0),'mtp':('on',0),'ngram':('off',3),'mtp-ngram':('on',3)}
+
+def save(p,j):
+    tmp=p.with_suffix('.tmp');tmp.write_text(json.dumps(j,indent=2)+'\n');tmp.replace(p)
+
+def option(args,key,value):
+    if key in args:args[args.index(key)+1]=str(value)
+    else:args.extend([key,str(value)])
+
+def first_difference(a,b):
+    return next((i for i,(x,y) in enumerate(zip(a,b)) if x!=y),
+                min(len(a),len(b)) if len(a)!=len(b) else None)
+
+def worker(out):
+    sys.path.insert(0,str(R/'tools'))
+    from configure_rtxpro import configure
+    out.mkdir(parents=True,exist_ok=False)
+    state={'started':time.time(),'source':(R/'source-commit.txt').read_text().strip(),
+        'kv':'fp16','weights':'Q8_0','records':[],
+        'engine_sha256':hashlib.sha256((R/'build/strata').read_bytes()).hexdigest(),
+        'scope':'One storage change; unchanged cache capacity, PCIe split, routing and adaptive waits.'}
+    target=out/'matrix.json'; save(target,state)
+    env={k:v for k,v in os.environ.items() if not k.startswith('STRATA_')}
+
+    def command(label,args):
+        state['current']=label;save(target,state)
+        with (out/(label+'.log')).open('w') as log:
+            p=subprocess.run(list(map(str,args)),env=env,stdout=log,stderr=subprocess.STDOUT,timeout=1800)
+        if p.returncode:raise RuntimeError(label+' failed with exit '+str(p.returncode))
+
+    def case(label,mode,rotate,*,engine=R,prompt=65536,output=1024,gate=False,cases=('coding','editing')):
+        cfg=configure({'weights':'Q8_0','kv':'fp16','load_projection':False},engine,H,16384 if gate else 73728)
+        cfg['env'].update(STRATA_PLE_PREFAULT_THREADS='8',STRATA_ADAPT_NOWAIT='0',STRATA_EXCHANGE_ROTATE=str(rotate))
+        option(cfg['args'],'--pcie-frac',-1)
+        option(cfg['args'],'--expert-cache',16000 if gate else 16192 if MODES[mode][0]=='on' else 16400)
+        config=out/(label+'-config.json');save(config,cfg)
+        cmd=[PY,R/'tools/bench_mtp_modes.py','--config',config,'--output',out/label,
+            '--input-tokens',str(prompt),'--output-tokens',str(output),'--mode',MODES[mode][0],
+            '--suffix-draft',str(MODES[mode][1]),'--verify-window','8','--mtp-window','4',
+            '--workload','long','--repetitions','1','--cases',*cases,
+            '--source-commit',state['source'] if engine==R else '080891d445d1f28eb244d4ce4fbce22942dc0b29']
+        started=time.time();command(label,cmd)
+        data=json.loads((out/label/'result.json').read_text());run=data['runs'][0]
+        log=(out/label/f'engine-mtp-{MODES[mode][0]}.log').read_text()
+        counters=[tuple(map(int,m)) for m in re.findall(r'exchange rotation: (\d+) blocks, (\d+) host memcpy bytes avoided',log)]
+        if rotate and (not counters or counters[-1][0]==0 or 'exchange buffer rotation enabled' not in log):
+            raise RuntimeError(label+': rotation not exercised')
+        rec={'label':label,'mode':mode,'rotate':rotate,'started':started,'finished':time.time(),
+            'engine_info':run['engine_info'],'startup_seconds':run['startup_seconds'],'cases':run['cases'],
+            'prompt_sha256':data['prompt_sha256'],'rotation_counters':counters,
+            'diagnostics':[x for x in log.splitlines() if any(k in x for k in
+                ('resident RAM:','expert tiers:','exchange rotation:','decode timing:'))]}
+        assert run['engine_info']['kv']=='fp16'
+        state['records'].append(rec);save(target,state);return rec
+
+    def compare(a,b):
+        assert a['prompt_sha256']==b['prompt_sha256']
+        assert a['engine_info']['expert_slots']==b['engine_info']['expert_slots']
+        rows=[]
+        for x,y in zip(a['cases'],b['cases']):
+            rows.append({'task':x['task'],'control_tps':x['decode_tps'],'rotation_tps':y['decode_tps'],
+                'gain_pct':100*(y['decode_tps']/x['decode_tps']-1),
+                'first_token_difference':first_difference(x['token_ids'],y['token_ids'])})
+        state.setdefault('comparisons',[]).append({'control':a['label'],'candidate':b['label'],'cases':rows})
+        save(target,state);return rows
+
+    try:
+        command('source-api-fixture',[R/'build/file_expert_source_test','--rotation-gpu'])
+        command('source-api-memcheck',['compute-sanitizer','--tool','memcheck','--error-exitcode','71',
+                R/'build/file_expert_source_test','--rotation-gpu'])
+        state['source_fixture_and_gpu_memcheck_passed']=True;save(target,state)
+        old=case('default-off-old','serial',0,engine=CONTROL,prompt=8192,output=256,gate=True,cases=('coding',))
+        new=case('default-off-new','serial',0,prompt=8192,output=256,gate=True,cases=('coding',))
+        state['default_off_exact_tokens']=old['cases'][0]['token_ids']==new['cases'][0]['token_ids'];save(target,state)
+        if not state['default_off_exact_tokens']:raise RuntimeError('Default-off output regression')
+        promising=[]
+        for index,mode in enumerate(MODES):
+            records={}
+            for rotate in ((0,1) if index%2==0 else (1,0)):
+                records[rotate]=case(f'{mode}-rotate{rotate}-r1',mode,rotate)
+            rows=compare(records[0],records[1])
+            if any(x['first_token_difference'] is not None for x in rows):
+                state.setdefault('semantic_gate_failures',[]).append(mode);save(target,state)
+                # Other decoding paths are independent; preserve evidence, do not bless this one.
+                continue
+            if max(x['gain_pct'] for x in rows)>=3:promising.append((mode,index))
+        for mode,index in promising:
+            records={}
+            for rotate in ((1,0) if index%2==0 else (0,1)):
+                records[rotate]=case(f'{mode}-rotate{rotate}-r2',mode,rotate)
+            rows=compare(records[0],records[1])
+            if any(x['first_token_difference'] is not None for x in rows):
+                state.setdefault('semantic_gate_failures',[]).append(mode+'-repeat')
+        state['completed']=True
+    except BaseException as error:state['error']=repr(error);raise
+    finally:state['finished']=time.time();save(target,state)
+
+def main():
+    if len(sys.argv)>1 and sys.argv[1]=='--worker':worker(Path(sys.argv[2]));return
+    sys.path.insert(0,str(H/'fleet-downloads'))
+    import run_q4_trace_1c0c2bb as base
+    source=(R/'source-commit.txt').read_text().strip()
+    base.wh.CUTOFF=time.time()+24*3600
+    r=base.CgroupRun(D,source)
+    r.s.update(shutdown_afterwards=False,operational_guard_not_user_deadline=True,
+        cutoff_utc=datetime.datetime.fromtimestamp(base.wh.CUTOFF,datetime.timezone.utc).isoformat(),
+        current='waiting for previously queued Q8 tests',dependency=str(PRIOR));r.save()
+    lock=(H/'fleet-downloads/.rtxpro-bandwidth.lock').open('a')
+    try:
+        while not PRIOR.exists() or not json.loads(PRIOR.read_text()).get('finished'):
+            r.check_time();time.sleep(5)
+        while True:
+            try:fcntl.flock(lock.fileno(),fcntl.LOCK_EX|fcntl.LOCK_NB);break
+            except BlockingIOError:r.check_time();time.sleep(5)
+        r.cpu('component-build',['g++','-std=c++17','-O1','-g','-fsanitize=address,undefined','-fno-omit-frame-pointer',
+            '-I'+str(R/'include'),R/'tests/core/exchange_storage_test.cpp','-o',R/'exchange-storage-sanitized'])
+        r.cpu('component-sanitizers',[R/'exchange-storage-sanitized'])
+        r.cpu('configure',['cmake','-S',R,'-B',R/'build','-G','Ninja','-DCMAKE_BUILD_TYPE=Release',
+            '-DSTRATA_ENABLE_CUDA=ON','-DCMAKE_CUDA_COMPILER=/usr/local/cuda/bin/nvcc',
+            '-DCUDAToolkit_ROOT=/usr/local/cuda','-DCMAKE_CUDA_FLAGS_RELEASE=-O3 -DNDEBUG',
+            '-DCMAKE_CUDA_ARCHITECTURES=120','-DSTRATA_BUILD_TESTS=OFF',
+            '-DSTRATA_MMQ_KQUANTS=ON','-DSTRATA_NATIVE_EXPERTS=ON','-DSTRATA_FLEET_CUDA_TRACE=ON',
+            '-DSTRATA_GGML_DIR='+str(H/'src/Strata/third_party/llama.cpp')])
+        r.cpu('build',['cmake','--build',R/'build','--target','strata','exchange_storage_test','file_expert_source_test','-j','8'])
+        r.s['engine_sha256']=hashlib.sha256((R/'build/strata').read_bytes()).hexdigest();r.save()
+        step=r.gpu('exchange-rotation',[PY,Path(__file__),'--worker',D/'{attempt}'],{},timeout=21600)
+        r.s['matrix']=str(D/step['label']/'matrix.json');r.save()
+    except BaseException as error:r.finish(error);raise
+    else:r.finish()
+
+if __name__=='__main__':main()