fix: publish rotating expert ownership atomically for prefetch readers

CMakeLists.txt

@@ -355,6 +355,7 @@ if(STRATA_ENABLE_CUDA OR STRATA_ENABLE_HIP)
   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)
+    target_link_libraries(exchange_storage_test PRIVATE Threads::Threads)
     add_test(NAME exchange_storage_test COMMAND exchange_storage_test)
     set_tests_properties(exchange_storage_test PROPERTIES TIMEOUT 60)
   endif()

docs/Q8_EXCHANGE_ROTATION.md

@@ -17,7 +17,8 @@ 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
+expert ownership moves by atomic slot IDs into an immutable table of host and
+CUDA device addresses. Background router lookahead reads those IDs safely. The model
 weights, resident expert count, cache replacement policy and quantization stay
 the same. The GPU-to-host eviction transfer is still required.
 
@@ -60,7 +61,8 @@ close/reopen and byte equality. Run under Compute Sanitizer memcheck.
 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.
+   1,024 output**, 73,728 allocation, FP16 KV, coding then editing requests with fresh prompt state. The adaptive expert cache
+   persists between those two requests, identically in the comparison arms.
 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

include/strata/core/exchange_storage.hpp

@@ -1,6 +1,9 @@
 #pragma once
 
+#include <atomic>
 #include <cstddef>
+#include <memory>
+#include <utility>
 #include <cstdint>
 #include <limits>
 #include <string>
@@ -9,8 +12,8 @@
 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.
+// owners until FileExpertSource::close(). Ownership changes via slot IDs; the
+// physical buffer addresses, bytes and CUDA aliases stay in place. This first path requires uniform, fully mapped/pinned expert slots.
 class ExchangeStorage {
 public:
     struct View {
@@ -29,43 +32,63 @@ public:
             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);
+        const size_t resident_slots = (size_t)(resident_bytes / blob_bytes);
+        if (resident_slots > std::numeric_limits<size_t>::max() - count) {
+            error = "exchange slot count overflow"; return false;
+        }
+        auto experts = std::make_unique<std::atomic<size_t>[]>(offsets.size());
+        std::vector<size_t> spares(count);
+        std::vector<View> buffers(resident_slots + count);
+        for (size_t slot = 0; slot < resident_slots; ++slot)
+            buffers[slot] = {resident_host + slot * blob_bytes, resident_device + slot * blob_bytes};
+        std::vector<bool> used(resident_slots, false);
         for (size_t i = 0; i < offsets.size(); ++i) {
             const uint64_t at = offsets[i];
+            experts[i].store(kAbsent, std::memory_order_relaxed);
             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};
+            experts[i].store((size_t)(at / blob_bytes), std::memory_order_relaxed);
         }
         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);
+        for (size_t q = 0; q < count; ++q) {
+            spares[q] = resident_slots + q;
+            buffers[spares[q]] = {exchange_host + q * blob_bytes, exchange_device + q * blob_bytes};
+        }
+        experts_ = std::move(experts);
+        expert_count_ = offsets.size();
         spares_.swap(spares);
+        buffers_.swap(buffers);
         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{};
+        if (expert >= expert_count_) return {};
+        const size_t slot = experts_[expert].load(std::memory_order_acquire);
+        return slot != kAbsent ? buffers_[slot] : View{};
     }
-    View spare(size_t q) const { return q < spares_.size() ? spares_[q] : View{}; }
+    View spare(size_t q) const { return q < spares_.size() ? buffers_[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] = {};
+        if (in >= expert_count_ || out >= expert_count_ || q >= spares_.size() ||
+            !staged || bytes != blob_bytes_) return false;
+        const size_t incoming_slot = experts_[in].load(std::memory_order_acquire);
+        if (incoming_slot == kAbsent || experts_[out].load(std::memory_order_acquire) != kAbsent ||
+            buffers_[spares_[q]].host != staged) return false;
+        // A background router lookahead can query residency to skip redundant
+        // file prefetches. Publish one atomic slot ID; the pointer/alias pair is
+        // immutable. Actual compute readers still obey the completion contract.
+        experts_[out].store(spares_[q], std::memory_order_release);
+        spares_[q] = incoming_slot;
+        experts_[in].store(kAbsent, std::memory_order_release);
         ++exchanges_;
         avoided_bytes_ += bytes;
         return true;
@@ -73,10 +96,17 @@ public:
 
     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; }
+    void clear() {
+        experts_.reset(); expert_count_ = 0; buffers_.clear(); spares_.clear();
+        blob_bytes_ = 0; exchanges_ = avoided_bytes_ = 0;
+    }
 
 private:
-    std::vector<View> experts_, spares_;
+    static constexpr size_t kAbsent = std::numeric_limits<size_t>::max();
+    std::unique_ptr<std::atomic<size_t>[]> experts_;
+    size_t expert_count_ = 0;
+    std::vector<View> buffers_; // immutable physical addresses after initialize
+    std::vector<size_t> spares_; // touched only by the serialized cache manager
     size_t blob_bytes_ = 0;
     uint64_t exchanges_ = 0, avoided_bytes_ = 0;
 };

tests/core/exchange_storage_test.cpp

@@ -5,6 +5,7 @@
 #include <iostream>
 #include <random>
 #include <set>
+#include <thread>
 #include <stdexcept>
 #include <vector>
 
@@ -33,6 +34,29 @@ void exercise(size_t bytes, size_t rounds) {
     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);
+    // A lookahead thread may inspect residency while ownership is committed.
+    // It only observes immutable address pairs, never reads mutable expert bytes.
+    std::atomic<bool> stop{false}, reader_failed{false};
+    std::atomic<uint64_t> observations{0};
+    std::vector<ExchangeStorage::View> valid;
+    for (size_t i = 0; i < n; ++i) if (storage.resident(i).host) valid.push_back(storage.resident(i));
+    for (size_t q = 0; q < spare_count; ++q) valid.push_back(storage.spare(q));
+    std::thread reader([&] {
+        while (!stop.load()) {
+            for (size_t i = 0; i < n; ++i) {
+                const auto view = storage.resident(i);
+                if (view.host && std::none_of(valid.begin(), valid.end(), [&](auto v) {
+                        return v.host == view.host && v.device == view.device; })) reader_failed.store(true);
+            }
+            ++observations;
+        }
+    });
+    struct Join {
+        std::atomic<bool>& stop; std::thread& reader;
+        void finish() { stop.store(true); if (reader.joinable()) reader.join(); }
+        ~Join() { finish(); }
+    } join{stop, reader};
+    while (!observations.load()) std::this_thread::yield();
     uint64_t exchanges = 0;
     for (size_t round = 0; round < rounds; ++round) {
         std::vector<size_t> ins, outs;
@@ -79,6 +103,8 @@ void exercise(size_t bytes, size_t rounds) {
             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");
+    join.finish();
+    require(!reader_failed.load() && observations.load(), "concurrent residency reader observed invalid alias");
     storage.clear();
     require(!storage.active() && !storage.resident(0).host && !storage.spare(0).host &&
             !storage.exchanges() && !storage.avoided_bytes(), "clear retained state");

tools/run_q8_exchange_rotation_fleet.py

@@ -130,7 +130,7 @@ def main():
         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',
+        r.cpu('component-build',['g++','-std=c++17','-pthread','-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',