feat(server): carry drafter feature ring in concurrent prefix checkpoints

server/docs/ENVIRONMENT.md

@@ -77,6 +77,7 @@ consolidation of this list into CLI flags is tracked as follow-up work.
 | `DFLASH_MMID_TELEMETRY` | unset | DEBUG: report MUL_MAT_ID dispatch, MMVQ variant, and per-node graph compatibility. |
 | `DFLASH_KVFLASH` | unset | Prefer the CLI: `--kvflash` (token count or `auto`). |
 | `DFLASH_PREFIX_CACHE_SLOTS` | 32 | Container-entrypoint equivalent of `--prefix-cache-slots`; not read directly by the native binary. |
+| `DFLASH_PREFIX_NO_FEAT` | unset | DEBUG/A-B: =1 drops the drafter feature-ring slab from concurrent prefix checkpoints (speculative `--max-concurrency` only) to measure the payload's cost and exercise the cold-ring restore path. |
 | `DFLASH_PREFILL_CACHE_SLOTS` | 0 | Container-entrypoint equivalent of `--prefill-cache-slots`; not read directly by the native binary. |
 | `DFLASH_PREFILL_POOL_TRIM_TOKENS` | unset | OPT-IN: trim cached allocations from legacy CUDA/HIP device pools at completed Qwen3.5 prefill chunk boundaries after each configured token interval. Intended for long, shape-changing prefills on non-VMM devices; each trim synchronizes the target backend and retires captured graphs. |
 | `DFLASH_SPLIT_FAST_ROLLBACK` | unset | OPT-IN: exact F32 checkpoints and replay-free rollback for local qwen35 target layer splits. Prefer `--target-split-fast-rollback`; adds checkpoint VRAM (~1.65 GiB for the measured Qwen3.6-27B q=16 split). |
@@ -292,6 +293,7 @@ consolidation of this list into CLI flags is tracked as follow-up work.
 - `DFLASH_PREFILL_POOL_TRIM_TOKENS` - qwen35_backend.cpp (OPT-IN: trim legacy device pools during long prefills)
 - `DFLASH_PREFILL_TIMING` - qwen35_backend.cpp (DEBUG: per-ubatch prefill build/alloc/compute timing)
 - `DFLASH_PREFIX_CACHE_SLOTS` - scripts/entrypoint.sh (maps to `--prefix-cache-slots`)
+- `DFLASH_PREFIX_NO_FEAT` - qwen35_seq_engine.cpp (DEBUG/A-B: =1 drops the drafter feature-ring slab from concurrent prefix checkpoints)
 - `DFLASH_QWEN35MOE_CACHE_SLOTS` - qwen35moe_backend.cpp
 - `DFLASH_QWEN35MOE_HOTNESS` - qwen35moe_backend.cpp
 - `DFLASH_QWEN35MOE_NEXT_PLACEMENT_OUT` - qwen35moe_backend.cpp

server/docs/PREFIX_CACHE.md

@@ -221,6 +221,25 @@ resident committed checkpoint buffers. During an atomic replacement, the new
 buffer and the selected victim can coexist briefly, so transient process memory
 can exceed the limit by up to one checkpoint.
 
+When the engine runs speculative continuous batching (a local same-device
+DFlash drafter under `--max-concurrency`), eligible checkpoints additionally
+carry the slot's drafter feature-ring slab so a restored prefix keeps the
+draft window warm — the resident estimate charges for it. AR-only engines
+keep the leaner KV + recurrent payload. A capture is ineligible while its
+slot is still floored by a featureless restore (below), and the bench/debug
+switch `DFLASH_PREFIX_NO_FEAT=1` drops the slab entirely (see
+ENVIRONMENT.md).
+
+A checkpoint without a feature payload still restores correctly under
+speculation. The engine then treats ring rows below the restore cut as
+untrusted: the drafter rebuilds its K/V window only from rows the
+occupying sequence wrote itself (the restore cut plus everything
+generated afterward), so proposals are never conditioned on a previous
+slot occupant's features. Such a slot also refrains from blessing those
+rows into later feature-bearing checkpoints — until it has written a full
+ring worth of positions past the restore cut, after which every row is
+self-written and the payload resumes.
+
 | Scenario | Typical prefix length | Recommended cap |
 |----------|----------------------|-----------------|
 | Single-user chat | 200–2000 tokens | 16–32 |

server/src/internal.h

@@ -623,6 +623,11 @@ struct PrefixSnapshot {
     //     [HEAD_DIM, kv_end-kv_start, N_HEAD_KV] (smaller than cache).
     //   - ssm_state_snap, conv_state_snap, target_feat_snap are NOT
     //     allocated (THIN snapshots are KV-only).
+    // For Layout::paged:
+    //   - target_feat_snap is optional. Speculating engines capture the
+    //     sequence slot's live drafter feature-ring slab so a restored prefix
+    //     keeps the draft window warm; engines without speculation leave it
+    //     null and pay nothing.
 };
 
 // Snapshot the slim state of `cache` into `snap`. KV tensors are RIGHT-SIZED
@@ -647,21 +652,28 @@ void free_prefix_snapshot(PrefixSnapshot & snap);
 // Exact CPU-buffer allocation size for the dense checkpoint layout used by
 // snapshot_paged_target_cache(). Returns zero when the cache topology or token
 // count is invalid. This lets the scheduler enforce a resident-memory budget
-// before allocating or copying a checkpoint.
+// before allocating or copying a checkpoint. `with_target_feat` must match the
+// capture call so the estimate covers the optional drafter feature payload.
 size_t estimate_paged_target_cache_snapshot_bytes(
-    const TargetCache & cache, int token_count);
+    const TargetCache & cache, int token_count, bool with_target_feat = false);
 
 // Capture one live sequence from a multi-slot paged cache. Attention rows are
 // gathered through `block_table` into dense logical order in the copied
 // snapshot; recurrent state is copied only from `seq_slot`'s slab. The page
 // table itself is intentionally not retained: every restore owns fresh pages.
+// When `with_target_feat` is set and the cache owns a drafter feature ring,
+// the slot's live ring slab is captured verbatim (ring slots are absolute
+// positions mod cap, so the first min(token_count, cap) slab rows are
+// self-describing on restore). Engines without speculation pass false and
+// keep the checkpoint at KV + recurrent state only.
 bool snapshot_paged_target_cache(
     const TargetCache & cache,
     int seq_slot,
     const std::vector<uint32_t> & block_table,
     int block_size,
     int token_count,
-    PrefixSnapshot & snap);
+    PrefixSnapshot & snap,
+    bool with_target_feat = false);
 
 // Atomically replace a paged snapshot. The incumbent remains valid when
 // allocation, layout validation, or any staged copy fails.
@@ -671,11 +683,14 @@ bool replace_paged_target_cache(
     const std::vector<uint32_t> & block_table,
     int block_size,
     int token_count,
-    PrefixSnapshot & destination);
+    PrefixSnapshot & destination,
+    bool with_target_feat = false);
 
 // Restore a copied paged snapshot into fresh destination pages and one
 // recurrent-state slab. `block_table` describes the destination sequence and
-// must cover snap.cur_pos logical tokens.
+// must cover snap.cur_pos logical tokens. A checkpoint that carries a
+// drafter feature payload also restores the slot's feature-ring slab; a
+// checkpoint without one restores KV + recurrent state only.
 bool restore_paged_target_cache(
     const PrefixSnapshot & snap,
     TargetCache & cache,

server/src/qwen35/concurrency/qwen35_seq_engine.cpp

@@ -11,6 +11,7 @@
 #include "graph_builders.h"
 #include "attn_masks.h"
 #include "prefill_helpers.h"
+#include "common/chain_rollback_policy.h"
 #include "common/concurrency/chain_spec_shapes.h"
 #include "common/dflash2_head.h"
 #include "common/sampler.h"
@@ -64,6 +65,7 @@ Qwen35SeqEngine::Qwen35SeqEngine(
     slot_draft_kv_.resize(static_cast<size_t>(n_slots));
     seq_lens_.assign(static_cast<size_t>(n_slots), 0);
     reserve_growth_.assign(static_cast<size_t>(n_slots), 0);
+    slot_ring_valid_from_.assign(static_cast<size_t>(n_slots), 0);
 
     fixed_chain_ready_ = fixed_chain_.enabled && fixed_chain_.width > 1 &&
         fixed_chain_.width <= 16 &&
@@ -264,6 +266,13 @@ Qwen35SeqEngine::prepare_chain_drafts(
             return std::nullopt;
         }
         lanes.push_back({i, input.slot, input.token, state, mirror});
+        // Ring floor from a featureless restore: rows below it belong to a
+        // previous occupant, so seed the append cursor past them — they stay
+        // unappended (slot_pos -1) and masked out of the draft context.
+        const int32_t ring_floor =
+            input.slot < static_cast<int>(slot_ring_valid_from_.size())
+                ? slot_ring_valid_from_[static_cast<size_t>(input.slot)] : 0;
+        if (state->next_pos < ring_floor) state->next_pos = ring_floor;
         if (!draft_kv_begin_step(
                 *state, b_.dw_, b_.draft_backend_, *mirror,
                 slots_.slot(input.slot).cur_pos)) {
@@ -380,17 +389,37 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit(
     AdmitResult result = slots_.admit(request_id, prompt, sampler);
     if (result.status == AdmitResult::Status::admitted) {
         reset_recurrent_slot(b_.cache_, result.slot);
-        if (result.slot >= 0 &&
-            result.slot < static_cast<int>(slot_draft_kv_.size()) &&
-            slot_draft_kv_[static_cast<size_t>(result.slot)]) {
-            draft_kv_reset(*slot_draft_kv_[static_cast<size_t>(result.slot)]);
-        }
+        reset_slot_draft_state(result.slot);
     }
     return result;
 }
 
+void Qwen35SeqEngine::reset_slot_draft_state(int slot) {
+    if (slot >= 0 && slot < static_cast<int>(slot_draft_kv_.size()) &&
+        slot_draft_kv_[static_cast<size_t>(slot)]) {
+        draft_kv_reset(*slot_draft_kv_[static_cast<size_t>(slot)]);
+    }
+    if (slot >= 0 && slot < static_cast<int>(slot_ring_valid_from_.size())) {
+        slot_ring_valid_from_[static_cast<size_t>(slot)] = 0;
+    }
+}
+
+// DFLASH_PREFIX_NO_FEAT=1 drops the drafter feature slab from concurrent
+// prefix checkpoints — a bench/debug knob that measures the payload's
+// contribution and exercises the cold-ring restore path.
+static bool prefix_feat_payload_disabled() {
+    static const bool off = env_flag_enabled("DFLASH_PREFIX_NO_FEAT");
+    return off;
+}
+
 size_t Qwen35SeqEngine::estimate_prefix_store_bytes(int tokens) const {
-    return estimate_paged_target_cache_snapshot_bytes(b_.cache_, tokens);
+    // Speculating engines also carry the slot's drafter feature-ring slab in
+    // each checkpoint; the estimate must charge for it so the resident-byte
+    // budget stays honest.
+    const bool with_target_feat =
+        fixed_chain_ready_ && !prefix_feat_payload_disabled();
+    return estimate_paged_target_cache_snapshot_bytes(
+        b_.cache_, tokens, with_target_feat);
 }
 
 int Qwen35SeqEngine::checkpoint_index(PrefixStoreRef checkpoint) const {
@@ -428,6 +457,12 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit_with_prefix(
 
     const int slot = result.slot;
     slots_.slot(slot).pending_capture = {};
+    // A restored prefix repopulates the slot's target_feat slab below; a
+    // cold admission leaves whatever the prefill graph will write. Either
+    // way the previous occupant's drafter K/V window must not survive —
+    // its next_pos/slot_pos bookkeeping would otherwise feed stale rows to
+    // the new sequence's drafts.
+    reset_slot_draft_state(slot);
     bool restored = false;
     if (plan.restore.valid()) {
         const int restore_index = checkpoint_index(plan.restore);
@@ -464,17 +499,30 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit_with_prefix(
             result.prefix_store.invalidated = plan.restore;
             if (result.status == AdmitResult::Status::admitted) {
                 reset_recurrent_slot(b_.cache_, result.slot);
+                reset_slot_draft_state(result.slot);
             } else {
                 result.error =
                     "cold admission failed after stale prefix restore";
             }
         } else {
             result.prefix_store.restored = plan.restore;
+            // A checkpoint without the feature slab leaves ring rows below
+            // the restore cut holding a previous occupant's features. Floor
+            // them so the drafter never bulk-appends rows this sequence did
+            // not write (unpopulated slots stay masked out of the draft
+            // context) and a later capture does not bless them.
+            if (snap && !snap->target_feat_snap &&
+                slot < static_cast<int>(slot_ring_valid_from_.size())) {
+                slot_ring_valid_from_[static_cast<size_t>(slot)] =
+                    plan.restore.tokens;
+            }
             std::fprintf(stderr,
                 "[parallel-pc] restored checkpoint=%llu seq_slot=%d "
-                "tokens=%d time_ms=%.1f\n",
+                "tokens=%d feat=%d time_ms=%.1f\n",
                 (unsigned long long)plan.restore.id, slot,
-                plan.restore.tokens, (double)restore_elapsed_us / 1000.0);
+                plan.restore.tokens,
+                snap && snap->target_feat_snap ? 1 : 0,
+                (double)restore_elapsed_us / 1000.0);
         }
         result.prefix_store.restore_attempted = true;
         result.prefix_store.restore_elapsed_us = restore_elapsed_us;
@@ -514,9 +562,22 @@ PrefixStoreEvent Qwen35SeqEngine::capture_prefix(
     }
     const auto capture_started = std::chrono::steady_clock::now();
     PrefixSnapshot & snapshot = b_.prefix_snapshots_[checkpoint];
+    // A slot whose ring still holds foreign rows below a featureless restore
+    // must not bless them into a feature payload. Ring rows are keyed by
+    // position % cap, so the foreign region is fully overwritten once the
+    // sequence has written `cap` positions past the cut and the payload
+    // resumes.
+    const int32_t ring_floor =
+        slot < static_cast<int>(slot_ring_valid_from_.size())
+            ? slot_ring_valid_from_[static_cast<size_t>(slot)] : 0;
+    const bool with_target_feat =
+        fixed_chain_ready_ && !prefix_feat_payload_disabled() &&
+        (ring_floor == 0 ||
+         ticket.checkpoint.tokens >= ring_floor + b_.cache_.target_feat_cap);
     if (!replace_paged_target_cache(
             b_.cache_, slot, sequence.block_table,
-            (int)pool_.block_size(), ticket.checkpoint.tokens, snapshot)) {
+            (int)pool_.block_size(), ticket.checkpoint.tokens, snapshot,
+            with_target_feat)) {
         event.elapsed_us =
             (uint64_t)std::chrono::duration_cast<std::chrono::microseconds>(
                 std::chrono::steady_clock::now() - capture_started).count();
@@ -1622,10 +1683,7 @@ bool Qwen35SeqEngine::restore_kv(int slot, std::string & error) {
         // state together — so it must be reset exactly as at admission.
         if (!slots_.resume_recompute(slot)) return false;
         reset_recurrent_slot(b_.cache_, slot);
-        if (slot < static_cast<int>(slot_draft_kv_.size()) &&
-            slot_draft_kv_[static_cast<size_t>(slot)]) {
-            draft_kv_reset(*slot_draft_kv_[static_cast<size_t>(slot)]);
-        }
+        reset_slot_draft_state(slot);
         return true;
     }
     std::vector<int32_t> blocks;
@@ -1651,10 +1709,7 @@ bool Qwen35SeqEngine::evict_kv(int slot, int32_t pending_token,
 void Qwen35SeqEngine::retire(int slot) {
     offload_.discard(slot);
     if (!slots_.is_active(slot)) return;
-    if (slot >= 0 && slot < static_cast<int>(slot_draft_kv_.size()) &&
-        slot_draft_kv_[static_cast<size_t>(slot)]) {
-        draft_kv_reset(*slot_draft_kv_[static_cast<size_t>(slot)]);
-    }
+    reset_slot_draft_state(slot);
     slots_.retire(slot);
 }
 

server/src/qwen35/concurrency/qwen35_seq_engine.h

@@ -168,6 +168,19 @@ class Qwen35SeqEngine final : public SeqEngine {
     bool arm_capture(
         int slot, PrefixCaptureTicket ticket, int restored_tokens);
     int checkpoint_index(PrefixStoreRef checkpoint) const;
+    // Invalidate the slot's drafter K/V window so the next speculative step
+    // bulk-appends from the feature ring. No-op when speculation is off.
+    void reset_slot_draft_state(int slot);
+
+    // Per-slot lower bound on feature-ring rows the occupying sequence wrote
+    // itself. A featureless checkpoint restore leaves ring rows below the
+    // restored cut holding a previous occupant's features; the floor keeps
+    // the drafter from bulk-appending them (they stay masked out of the
+    // draft context) and keeps captures from claiming them as a feature
+    // payload until the sequence has written cap positions past the cut and
+    // every ring row is self-written again. Zero means the whole ring slab
+    // is self-written.
+    std::vector<int32_t> slot_ring_valid_from_;
 
     PagedKvPool & pool_;
     Qwen35Backend & b_;