From ad6ccbad4f8bc21ed7885dd842365cb54533902f Mon Sep 17 00:00:00 2001
From: NeelM0906 <neelanjan.mitra@acti.ai>
Date: Thu, 20 Aug 2026 15:57:36 -0400
Subject: [PATCH 1/6] perf: coalesce disk-adjacent expert misses into scattered
 preadv runs
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Misses whose blobs are exactly contiguous on disk (uniform stride or a
permuted expertOffsets table) are grouped into runs and fetched with one
preadv per run — one syscall and one sequential NVMe read instead of a
random pread per expert. Applies to both the cache-plan path and the
speculative/async fill path. Roadmap Task 14e.
---
 .../Streaming/PreadExpertStreamer.swift       | 104 +++++++++++--
 .../PreadExpertStreamerTests+Coalescing.swift | 139 ++++++++++++++++++
 2 files changed, 231 insertions(+), 12 deletions(-)
 create mode 100644 Tests/Mference/Core/Infrastructure/Streaming/PreadExpertStreamerTests+Coalescing.swift

diff --git a/Sources/Mference/Infrastructure/Streaming/PreadExpertStreamer.swift b/Sources/Mference/Infrastructure/Streaming/PreadExpertStreamer.swift
index f3224b0..24d46ac 100644
--- a/Sources/Mference/Infrastructure/Streaming/PreadExpertStreamer.swift
+++ b/Sources/Mference/Infrastructure/Streaming/PreadExpertStreamer.swift
@@ -275,15 +275,23 @@ public final class PreadExpertStreamer: @unchecked Sendable {
         precondition(plan.assignedSlots.count == plan.experts.count,
                      "expert cache plan slot count mismatch")
 
+        let missFileOffsets = try plan.misses.map { index in
+            try fileOffsetForExpert(plan.experts[index])
+        }
+        let runs = Self.coalescedReadRuns(offsets: missFileOffsets,
+                                          stride: layout.expertStride)
         let errorLock = NSLock()
         nonisolated(unsafe) var firstError: Error?
-        DispatchQueue.concurrentPerform(iterations: plan.misses.count) { missOffset in
-            let index = plan.misses[missOffset]
+        DispatchQueue.concurrentPerform(iterations: runs.count) { runIndex in
+            let run = runs[runIndex]
+            let destinations = run.map {
+                self.slotPointers[plan.assignedSlots[plan.misses[$0]]]
+            }
             do {
-                _ = try self.loadExpert(
-                    layer: 0,
-                    expert: plan.experts[index],
-                    slot: plan.assignedSlots[index])
+                try self.readScattered(
+                    into: destinations,
+                    fileOffset: missFileOffsets[run[0]],
+                    strideBytes: Int(self.layout.expertStride))
             } catch {
                 errorLock.lock()
                 if firstError == nil { firstError = error }
@@ -378,15 +386,23 @@ public final class PreadExpertStreamer: @unchecked Sendable {
         _ reservation: [(expert: Int, slot: Int)]
     ) -> UInt64 {
         guard !reservation.isEmpty else { return 0 }
+        let fileOffsets = reservation.map { entry in
+            (try? fileOffsetForExpert(entry.expert)) ?? UInt64.max
+        }
+        let readable = reservation.indices.filter { fileOffsets[$0] != UInt64.max }
+        let runs = Self.coalescedReadRuns(offsets: readable.map { fileOffsets[$0] },
+                                          stride: layout.expertStride)
         let loadedLock = NSLock()
         nonisolated(unsafe) var loaded: [Int] = []
-        DispatchQueue.concurrentPerform(iterations: reservation.count) { index in
-            let entry = reservation[index]
-            guard (try? self.loadExpert(layer: 0,
-                                        expert: entry.expert,
-                                        slot: entry.slot)) != nil else { return }
+        DispatchQueue.concurrentPerform(iterations: runs.count) { runIndex in
+            let entries = runs[runIndex].map { readable[$0] }
+            let destinations = entries.map { self.slotPointers[reservation[$0].slot] }
+            guard (try? self.readScattered(
+                into: destinations,
+                fileOffset: fileOffsets[entries[0]],
+                strideBytes: Int(self.layout.expertStride))) != nil else { return }
             loadedLock.lock()
-            loaded.append(index)
+            loaded.append(contentsOf: entries)
             loadedLock.unlock()
         }
 
@@ -558,6 +574,70 @@ public final class PreadExpertStreamer: @unchecked Sendable {
             maxCallNanos: maxCallNanos)
     }
 
+    private func fileOffsetForExpert(_ expert: Int) throws -> UInt64 {
+        let regionOffset = layout.expertOffset(layer: 0, expert: expert)
+        guard regionOffset + layout.expertStride <= layout.streamSize else {
+            throw StreamerError.offsetOutOfRange(regionOffset)
+        }
+        return layout.streamOffset + regionOffset
+    }
+
+    /// Groups reads of `stride` bytes at `offsets` into runs that are exactly
+    /// contiguous on disk, so each run can be fetched with one scattered
+    /// `preadv` instead of one random `pread` per expert. Returns runs of
+    /// indices into `offsets`, each run in ascending disk order. Duplicate
+    /// offsets never share a run: their reads would overlap.
+    static func coalescedReadRuns(offsets: [UInt64], stride: UInt64) -> [[Int]] {
+        let sorted = offsets.indices.sorted { offsets[$0] < offsets[$1] }
+        var runs: [[Int]] = []
+        for index in sorted {
+            if let last = runs.last?.last, offsets[index] == offsets[last] &+ stride {
+                runs[runs.count - 1].append(index)
+            } else {
+                runs.append([index])
+            }
+        }
+        return runs
+    }
+
+    /// Reads `destinations.count * strideBytes` contiguous file bytes starting
+    /// at `fileOffset`, scattering `strideBytes` into each destination in
+    /// order. Single-destination runs use the plain `pread` path.
+    private func readScattered(into destinations: [UnsafeMutableRawPointer],
+                               fileOffset: UInt64,
+                               strideBytes: Int) throws {
+        guard destinations.count > 1 else {
+            return try readFull(into: destinations[0],
+                                fileOffset: fileOffset,
+                                count: strideBytes)
+        }
+        let total = destinations.count * strideBytes
+        var filled = 0
+        while filled < total {
+            let startIndex = filled / strideBytes
+            let within = filled % strideBytes
+            var vectors = [iovec(
+                iov_base: destinations[startIndex].advanced(by: within),
+                iov_len: strideBytes - within)]
+            for index in (startIndex + 1)..<destinations.count {
+                vectors.append(iovec(iov_base: destinations[index],
+                                     iov_len: strideBytes))
+            }
+            let readCount = vectors.withUnsafeBufferPointer { buffer in
+                preadv(fd, buffer.baseAddress, Int32(buffer.count),
+                       off_t(fileOffset) + off_t(filled))
+            }
+            if readCount < 0 {
+                throw StreamerError.preadFailed(errno: errno)
+            }
+            if readCount == 0 {
+                throw StreamerError.sizeMismatch(expected: UInt64(total),
+                                                 actual: UInt64(filled))
+            }
+            filled += readCount
+        }
+    }
+
     private func readFull(into destination: UnsafeMutableRawPointer,
                           fileOffset: UInt64,
                           count: Int) throws {
diff --git a/Tests/Mference/Core/Infrastructure/Streaming/PreadExpertStreamerTests+Coalescing.swift b/Tests/Mference/Core/Infrastructure/Streaming/PreadExpertStreamerTests+Coalescing.swift
new file mode 100644
index 0000000..c6b55f8
--- /dev/null
+++ b/Tests/Mference/Core/Infrastructure/Streaming/PreadExpertStreamerTests+Coalescing.swift
@@ -0,0 +1,139 @@
+import Darwin
+import Foundation
+import Metal
+import Testing
+
+@testable import Mference
+
+/// Coalesced miss reads: misses whose blobs are adjacent on disk are fetched
+/// with one scattered `preadv` per contiguous run instead of one `pread` per
+/// expert. Correctness must be identical for uniform layouts, permuted
+/// `expertOffsets` tables, and the speculative fill path.
+extension PreadExpertStreamerTests {
+
+  @Test func coalescedReadRuns_groupsContiguousOffsets() {
+    let stride = UInt64(Self.expertStride)
+    // Offsets 0,1s,2s are one run; 4s,5s another; 7s alone. Input order is
+    // scrambled to prove the grouping sorts by disk position first.
+    let offsets: [UInt64] = [4 * stride, 0, 2 * stride, 7 * stride, stride, 5 * stride]
+    let runs = PreadExpertStreamer.coalescedReadRuns(offsets: offsets, stride: stride)
+    #expect(runs == [[1, 4, 2], [0, 5], [3]])
+  }
+
+  @Test func coalescedReadRuns_duplicateOffsetsNeverShareARun() {
+    let stride = UInt64(Self.expertStride)
+    let offsets: [UInt64] = [0, 0, stride]
+    let runs = PreadExpertStreamer.coalescedReadRuns(offsets: offsets, stride: stride)
+    // Overlapping reads must stay separate syscalls; only one duplicate can
+    // extend into the following contiguous blob.
+    #expect(runs.count == 2)
+    #expect(runs.flatMap { $0 }.sorted() == [0, 1, 2])
+    for run in runs {
+      let sortedOffsets = run.map { offsets[$0] }
+      #expect(sortedOffsets == sortedOffsets.sorted())
+      for pair in zip(sortedOffsets, sortedOffsets.dropFirst()) {
+        #expect(pair.1 == pair.0 + stride)
+      }
+    }
+  }
+
+  @Test func cachePlanWithAdjacentMisses_readsEverySlotCorrectly() throws {
+    let url = try Self.writeSyntheticLayer()
+    defer { try? FileManager.default.removeItem(at: url) }
+    let device = try MetalContext().device
+    let streamer = try PreadExpertStreamer(
+      layout: Self.makeLayout(path: url.path), device: device,
+      slotCount: Self.numExperts)
+
+    // Cold cache: all four experts miss and are contiguous on disk, so this
+    // exercises a single multi-entry scattered read.
+    let results = try streamer.loadExpertsCached(experts: Array(0..<Self.numExperts))
+    for (expert, r) in results.enumerated() {
+      let got = Self.bytes(of: r.buffer, offset: r.offset, count: Self.expertStride)
+      #expect(
+        got.allSatisfy { $0 == Self.tagByte(expert) },
+        "expert \(expert) slot not uniformly tagged after coalesced read")
+    }
+  }
+
+  @Test func cachePlanWithPermutedOffsetTable_readsEverySlotCorrectly() throws {
+    let url = try Self.writeSyntheticLayer()
+    defer { try? FileManager.default.removeItem(at: url) }
+    let device = try MetalContext().device
+    let stride = UInt64(Self.expertStride)
+    // Expert id -> disk position: id 0 lives at blob 2, id 1 at blob 3,
+    // id 2 at blob 0, id 3 at blob 1. Runs form on disk order (ids 2,3 then
+    // 0,1), not id order.
+    let positions = [2, 3, 0, 1]
+    let layout = StreamLayout(
+      path: url.path,
+      streamOffset: Self.streamOffset,
+      streamSize: Self.streamSize,
+      expertsPerLayer: Self.numExperts,
+      expertStride: stride,
+      expertOffsets: positions.map { UInt64($0) * stride })
+    let streamer = try PreadExpertStreamer(
+      layout: layout, device: device, slotCount: Self.numExperts)
+
+    let results = try streamer.loadExpertsCached(experts: Array(0..<Self.numExperts))
+    for (expert, r) in results.enumerated() {
+      let got = Self.bytes(of: r.buffer, offset: r.offset, count: Self.expertStride)
+      #expect(
+        got.allSatisfy { $0 == Self.tagByte(positions[expert]) },
+        "expert \(expert) should hold the blob at disk position \(positions[expert])")
+    }
+  }
+
+  @Test func speculativeFillWithAdjacentExperts_publishesTaggedSlots() throws {
+    let url = try Self.writeSyntheticLayer()
+    defer { try? FileManager.default.removeItem(at: url) }
+    let device = try MetalContext().device
+    let streamer = try PreadExpertStreamer(
+      layout: Self.makeLayout(path: url.path), device: device,
+      slotCount: Self.numExperts)
+
+    let reservation = streamer.reserveSpeculativeSlots(
+      experts: [1, 2], keepEvictable: 0)
+    #expect(reservation.count == 2)
+    let bytes = streamer.executeSpeculativeReservation(reservation)
+    #expect(bytes == UInt64(2 * Self.expertStride))
+
+    let resident = streamer.residentExpertsSnapshot()
+    for entry in reservation {
+      #expect(resident[entry.slot] == entry.expert)
+    }
+    // A follow-up plan must treat both as hits and read nothing.
+    let plan = streamer.planExpertsCached(experts: [1, 2])
+    #expect(plan.misses.isEmpty)
+    let results = try streamer.executeExpertCachePlan(plan)
+    for (index, expert) in [1, 2].enumerated() {
+      let got = Self.bytes(
+        of: results[index].buffer, offset: results[index].offset,
+        count: Self.expertStride)
+      #expect(got.allSatisfy { $0 == Self.tagByte(expert) })
+    }
+  }
+
+  @Test func coalescedPlanHittingEOF_throwsInsteadOfPublishing() throws {
+    let url = try Self.writeSyntheticLayer()
+    defer { try? FileManager.default.removeItem(at: url) }
+    let device = try MetalContext().device
+    let streamer = try PreadExpertStreamer(
+      layout: Self.makeLayout(path: url.path), device: device,
+      slotCount: Self.numExperts)
+
+    // Truncate mid-run: experts 2 and 3 are one contiguous run, but the file
+    // now ends inside expert 3's blob.
+    let truncatedLen =
+      off_t(Self.streamOffset) + off_t(3 * Self.expertStride + Self.expertStride / 2)
+    #expect(truncate(url.path, truncatedLen) == 0)
+
+    #expect(throws: StreamerError.self) {
+      _ = try streamer.loadExpertsCached(experts: [2, 3])
+    }
+    // The failed run must not publish either slot as resident.
+    let resident = streamer.residentExpertsSnapshot()
+    #expect(!resident.contains(3))
+  }
+
+}

From c669b0ad97e53a08a8db50c369752dbc223e35a6 Mon Sep 17 00:00:00 2001
From: NeelM0906 <neelanjan.mitra@acti.ai>
Date: Thu, 20 Aug 2026 16:08:20 -0400
Subject: [PATCH 2/6] perf: port pilot/shadow speculative expert prefetch to
 Inkling decode

SpeculativeRouterInkling mirrors the DSV4 pilot: layer L+1's sigmoid
router run against layer L's post-attention state inside layer L's own
command buffer, read at the same router wake, feeding the existing
reserve/read/join/confirm machinery and the shadow issue budget. Same
kernels as the real router, so ranking is bit-identical (pinned by a
parity test); a real-install test pins off-vs-shadow greedy identity.
Mode stays opt-in (MFERENCE_SPEC_PREFETCH) until an A/B on the install
accepts a default, matching how DSV4 earned its shadow default.
---
 .../MoE/SpeculativeRouterInkling.swift        | 110 ++++++++++++++++++
 .../Runtime/Inference/RealForwardRunner.swift |  73 +++++++++++-
 .../SpeculativeRouterInklingTests.swift       | 101 ++++++++++++++++
 .../InklingGenerationRegressionTests.swift    |  39 +++++++
 4 files changed, 321 insertions(+), 2 deletions(-)
 create mode 100644 Sources/Mference/Kernels/MoE/SpeculativeRouterInkling.swift
 create mode 100644 Tests/Mference/Core/Kernels/Inkling/SpeculativeRouterInklingTests.swift

diff --git a/Sources/Mference/Kernels/MoE/SpeculativeRouterInkling.swift b/Sources/Mference/Kernels/MoE/SpeculativeRouterInkling.swift
new file mode 100644
index 0000000..ad7bb63
--- /dev/null
+++ b/Sources/Mference/Kernels/MoE/SpeculativeRouterInkling.swift
@@ -0,0 +1,110 @@
+import Foundation
+import Metal
+
+/// PILOT router-lookahead for Inkling-Small: layer L+1's sigmoid router
+/// applied to layer L's post-attention state, encoded into layer L's command
+/// buffer so the CPU learns a *guess* at the next layer's expert set at the
+/// same wake as the real router readback — no extra command buffer, no extra
+/// sync. The Inkling twin of `SpeculativeRouterDSV4`.
+///
+/// Deliberately a separate object from `InklingKernels` rather than an extra
+/// method on it: it must not share that class's `routerLogits` singleton (the
+/// real router's logits are live in the same command buffer), and keeping it
+/// in its own file means the speculative path cannot perturb the validated
+/// decode router. Same kernels (`router_gemv_bf16_r4` +
+/// `inkling_router_select`, both built without function constants exactly as
+/// `InklingKernels` builds them), so the ranking — sigmoid scoring with the
+/// per-expert gate bias — is bit-identical to what layer L+1 will compute for
+/// real. Recall depends entirely on that fidelity; only the *input
+/// activation* is an approximation.
+final class SpeculativeRouterInkling {
+
+    private let gemvPSO: MTLComputePipelineState
+    private let selectPSO: MTLComputePipelineState
+    private let logits: MTLBuffer
+    /// Predicted expert ids, `topK` x UInt32. Read by the CPU on the router
+    /// signal.
+    let predictedIndices: MTLBuffer
+    /// The select kernel writes routing weights and shared-expert gammas
+    /// alongside the indices; the prediction never uses them, but the kernel
+    /// contract needs the buffers.
+    private let discardedWeights: MTLBuffer
+    private let discardedGammas: MTLBuffer
+
+    init(context: MetalContext,
+         numRouted: Int,
+         numShared: Int,
+         topK: Int) throws {
+        self.gemvPSO = try context.pipeline("router_gemv_bf16_r4")
+        self.selectPSO = try context.pipeline("inkling_router_select")
+
+        guard let logits = context.device.makeBuffer(
+            length: (numRouted + numShared) * MemoryLayout<Float>.stride,
+            options: .storageModeShared),
+            let indices = context.device.makeBuffer(
+                length: topK * MemoryLayout<UInt32>.stride,
+                options: .storageModeShared),
+            let weights = context.device.makeBuffer(
+                length: topK * MemoryLayout<Float16>.stride,
+                options: .storageModeShared),
+            let gammas = context.device.makeBuffer(
+                length: max(numShared, 1) * MemoryLayout<Float>.stride,
+                options: .storageModeShared) else {
+            throw MetalError.noDevice
+        }
+        logits.label = "inkling.pilot_router_logits"
+        indices.label = "inkling.pilot_router_indices"
+        self.logits = logits
+        self.predictedIndices = indices
+        self.discardedWeights = weights
+        self.discardedGammas = gammas
+    }
+
+    /// Encodes the lookahead GEMV + sigmoid top-k. Must be encoded *before*
+    /// the router signal so the CPU sees the prediction at the same wake as
+    /// the real indices. Dispatch geometry mirrors
+    /// `InklingKernels.encodeRouter` exactly.
+    func encodePrediction(commandBuffer cb: MTLCommandBuffer,
+                          weights: MTLBuffer, weightsOffset: Int,
+                          hidden: MTLBuffer,
+                          onesScale: MTLBuffer,
+                          gateBias: MTLBuffer, gateBiasOffset: Int,
+                          globalScale: MTLBuffer, globalScaleOffset: Int,
+                          numRouted: UInt32, numShared: UInt32,
+                          topK: UInt32, routeScale: Float,
+                          d: UInt32) {
+        var total = numRouted + numShared
+        var dim = d
+        if let enc = cb.makeComputeCommandEncoder() {
+            enc.setComputePipelineState(gemvPSO)
+            enc.setBuffer(weights, offset: weightsOffset, index: 0)
+            enc.setBuffer(hidden, offset: 0, index: 1)
+            enc.setBuffer(onesScale, offset: 0, index: 2)
+            enc.setBuffer(logits, offset: 0, index: 3)
+            enc.setBytes(&total, length: 4, index: 4)
+            enc.setBytes(&dim, length: 4, index: 5)
+            enc.dispatchThreadgroups(
+                MTLSize(width: (Int(total) + 3) / 4, height: 1, depth: 1),
+                threadsPerThreadgroup: MTLSize(width: 128, height: 1, depth: 1))
+            enc.endEncoding()
+        }
+
+        guard let enc = cb.makeComputeCommandEncoder() else { return }
+        var nr = numRouted, ns = numShared, tk = topK, rs = routeScale
+        enc.setComputePipelineState(selectPSO)
+        enc.setBuffer(logits, offset: 0, index: 0)
+        enc.setBuffer(gateBias, offset: gateBiasOffset, index: 1)
+        enc.setBuffer(globalScale, offset: globalScaleOffset, index: 2)
+        enc.setBuffer(predictedIndices, offset: 0, index: 3)
+        enc.setBuffer(discardedWeights, offset: 0, index: 4)
+        enc.setBuffer(discardedGammas, offset: 0, index: 5)
+        enc.setBytes(&nr, length: 4, index: 6)
+        enc.setBytes(&ns, length: 4, index: 7)
+        enc.setBytes(&tk, length: 4, index: 8)
+        enc.setBytes(&rs, length: 4, index: 9)
+        enc.dispatchThreadgroups(
+            MTLSize(width: 1, height: 1, depth: 1),
+            threadsPerThreadgroup: MTLSize(width: 32, height: 1, depth: 1))
+        enc.endEncoding()
+    }
+}
diff --git a/Sources/Mference/Runtime/Inference/RealForwardRunner.swift b/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
index 28111f6..17122c4 100644
--- a/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
+++ b/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
@@ -517,8 +517,12 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
     /// PILOT lookahead kernels, built on first use so `off` pays nothing and
     /// the (concurrently edited) DSV4 init block stays untouched.
     private var pilotRouter: SpeculativeRouterDSV4?
-    /// The prediction read out of `pilotRouter` (or the hash table) at the
-    /// current layer's router wake, consumed by `issueSpeculativePrefetch`.
+    /// Inkling twin of `pilotRouter`: same lazy construction, same lifecycle,
+    /// family-matched scoring kernels.
+    private var inklingPilotRouter: SpeculativeRouterInkling?
+    /// The prediction read out of `pilotRouter`/`inklingPilotRouter` (or the
+    /// hash table) at the current layer's router wake, consumed by
+    /// `issueSpeculativePrefetch`.
     private var pilotPrediction: (layer: Int, experts: [Int])?
 
     public init(model: Model, context: MetalContext, maxContext: Int,
@@ -1296,6 +1300,29 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
         return pilotRouter
     }
 
+    private func ensureInklingPilotRouter() -> SpeculativeRouterInkling? {
+        if let inklingPilotRouter { return inklingPilotRouter }
+        inklingPilotRouter = try? SpeculativeRouterInkling(
+            context: ctx,
+            numRouted: cfg.numExperts,
+            numShared: cfg.numSharedExperts,
+            topK: cfg.topKExperts)
+        return inklingPilotRouter
+    }
+
+    /// Inkling wake-side twin of `capturePilotPrediction`: no hash-routed
+    /// layers exist in this family, so the prediction comes only from the
+    /// pilot GEMV.
+    private func captureInklingPilotPrediction(nextLayer: Int, gemvEncoded: Bool) {
+        pilotPrediction = nil
+        guard speculativePrefetchMode == .pilot || speculativePrefetchMode == .shadow,
+              nextLayer < cfg.numLayers, gemvEncoded,
+              let buffer = inklingPilotRouter?.predictedIndices else { return }
+        let ptr = buffer.contents().assumingMemoryBound(to: UInt32.self)
+        let cap = cfg.numExperts - 1
+        pilotPrediction = (nextLayer, (0..<cfg.topKExperts).map { min(Int(ptr[$0]), cap) })
+    }
+
     /// Records this layer's routing for the next token's predictor.
     private func recordRoutedExperts(_ experts: [Int], layer: Int) {
         if previousTokenExperts.count != cfg.numLayers {
@@ -3710,6 +3737,36 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
                                  routeScale: Float(cfg.routedScalingFactor)
                                      / Self.inklingFFNPrescale,
                                  d: D)
+            // PILOT lookahead: layer L+1's router against *this* layer's
+            // post-attention state, into private buffers, before the signal —
+            // the CPU reads real L indices and speculative L+1 indices at one
+            // wake. Dense layers all precede the MoE layers in this family,
+            // so L+1 here is always routed (or past the last layer, which
+            // `shouldEncodePilotGemv` rejects).
+            var pilotGemvEncoded = false
+            if shouldEncodePilotGemv(nextLayer: L + 1),
+               let pilot = ensureInklingPilotRouter(),
+               let nextRouter = try? model.router(layer: L + 1),
+               let nextGateBias = try? model.inklingGateBias(layer: L + 1),
+               let nextGateScale = try? model.inklingGateGlobalScale(layer: L + 1) {
+                pilot.encodePrediction(
+                    commandBuffer: cb,
+                    weights: nextRouter.buffer,
+                    weightsOffset: Int(nextRouter.offset),
+                    hidden: routedX,
+                    onesScale: effectiveScaleBuffers[L + 1],
+                    gateBias: nextGateBias.buffer,
+                    gateBiasOffset: Int(nextGateBias.offset),
+                    globalScale: nextGateScale.buffer,
+                    globalScaleOffset: Int(nextGateScale.offset),
+                    numRouted: UInt32(cfg.numExperts),
+                    numShared: UInt32(cfg.numSharedExperts),
+                    topK: UInt32(cfg.topKExperts),
+                    routeScale: Float(cfg.routedScalingFactor)
+                        / Self.inklingFFNPrescale,
+                    d: D)
+                pilotGemvEncoded = true
+            }
             let routerSignal = encodeRouterSignal(cb)
             let sharedProjs = inklingSharedProjections[L]
             try! shared.encode(commandBuffer: cb,
@@ -3750,6 +3807,14 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
             for i in 0..<cfg.topKExperts {
                 experts[i] = min(Int(idxPtr[i]), cfg.numExperts - 1)
             }
+            // Join any speculative read aimed at this layer before planning —
+            // the plan must not evict a slot a background pread is filling —
+            // and score the prediction. Then this layer's routing becomes the
+            // next token's prediction for the same layer.
+            settleSpeculation(layer: L, actualExperts: experts)
+            recordRoutedExperts(experts, layer: L)
+            captureInklingPilotPrediction(nextLayer: L + 1,
+                                          gemvEncoded: pilotGemvEncoded)
             let topK = UInt32(cfg.topKExperts)
             let routedOffsets = model.routedExpertOffsets(layer: L)
             let plannedFetch = try model.planRoutedExperts(layer: L, experts: experts)
@@ -3855,6 +3920,10 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
                                               count: D,
                                               scale: Self.inklingFFNPrescale)
             routedCB.commit()
+            // The GPU is now busy with routed work; the CPU is free until the
+            // next layer's router wake — the window where a speculative read
+            // never competes with a critical-path pread.
+            issueSpeculativePrefetch(layer: L + 1)
             if Self.inklingDebugEnabled {
                 waitForCompletion(routedCB)
                 if !(L <= 6 || L == 41) {
diff --git a/Tests/Mference/Core/Kernels/Inkling/SpeculativeRouterInklingTests.swift b/Tests/Mference/Core/Kernels/Inkling/SpeculativeRouterInklingTests.swift
new file mode 100644
index 0000000..66f3440
--- /dev/null
+++ b/Tests/Mference/Core/Kernels/Inkling/SpeculativeRouterInklingTests.swift
@@ -0,0 +1,101 @@
+import Foundation
+import Metal
+import Testing
+
+@testable import Mference
+import MferenceValidationSupport
+
+/// The Inkling pilot router's whole value is ranking fidelity: it runs the
+/// same GEMV and select kernels as the real router, so on identical inputs
+/// its predicted expert ids must be bit-identical to the real readback.
+@Suite struct SpeculativeRouterInklingTests {
+
+    private static let numRouted = 256
+    private static let numShared = 2
+    private static let topK = 6
+    private static let d = 64
+
+    private static func bf16(_ v: Float) -> UInt16 {
+        UInt16(truncatingIfNeeded: v.bitPattern >> 16)
+    }
+
+    private static func buffer<T>(_ device: MTLDevice, _ values: [T]) -> MTLBuffer {
+        values.withUnsafeBytes { raw in
+            device.makeBuffer(bytes: raw.baseAddress!, length: raw.count,
+                              options: .storageModeShared)!
+        }
+    }
+
+    @Test func predictionMatchesRealRouterBitForBit() throws {
+        let context = try MetalContext()
+        let device = context.device
+        let kernels = try InklingKernels(context: context,
+                                         numRouted: Self.numRouted,
+                                         numShared: Self.numShared)
+        let pilot = try SpeculativeRouterInkling(context: context,
+                                                 numRouted: Self.numRouted,
+                                                 numShared: Self.numShared,
+                                                 topK: Self.topK)
+
+        var rng = SplitMix64(seed: 0x1_2C0F)
+        let total = Self.numRouted + Self.numShared
+        let weights = Self.buffer(
+            device,
+            (0..<total * Self.d).map { _ in Self.bf16(rng.uniform(0, 1) * 2 - 1) })
+        let hidden = Self.buffer(
+            device,
+            (0..<Self.d).map { _ in Float16(rng.uniform(0, 1) * 2 - 1) })
+        let onesScale = Self.buffer(
+            device,
+            (0..<Self.d).map { _ in Self.bf16(rng.uniform(0, 1) + 0.5) })
+        let gateBias = Self.buffer(
+            device,
+            (0..<total).map { _ in rng.uniform(0, 1) * 0.2 - 0.1 })
+        let globalScale = Self.buffer(device, [Float(1.25)])
+        let realIndices = Self.buffer(device,
+                                      [UInt32](repeating: .max, count: Self.topK))
+        let realWeights = Self.buffer(device,
+                                      [Float16](repeating: 0, count: 8))
+        let routeScale: Float = 1.7
+
+        let cb = context.queue.makeCommandBuffer()!
+        kernels.encodeRouter(commandBuffer: cb,
+                             weights: weights, weightsOffset: 0,
+                             hidden: hidden,
+                             onesScale: onesScale,
+                             gateBias: gateBias, gateBiasOffset: 0,
+                             globalScale: globalScale, globalScaleOffset: 0,
+                             outIndices: realIndices,
+                             outWeights: realWeights,
+                             numRouted: UInt32(Self.numRouted),
+                             numShared: UInt32(Self.numShared),
+                             topK: UInt32(Self.topK),
+                             routeScale: routeScale,
+                             d: UInt32(Self.d))
+        pilot.encodePrediction(commandBuffer: cb,
+                               weights: weights, weightsOffset: 0,
+                               hidden: hidden,
+                               onesScale: onesScale,
+                               gateBias: gateBias, gateBiasOffset: 0,
+                               globalScale: globalScale, globalScaleOffset: 0,
+                               numRouted: UInt32(Self.numRouted),
+                               numShared: UInt32(Self.numShared),
+                               topK: UInt32(Self.topK),
+                               routeScale: routeScale,
+                               d: UInt32(Self.d))
+        cb.commit()
+        cb.waitUntilCompleted()
+        #expect(cb.error == nil)
+
+        let real = realIndices.contents()
+            .bindMemory(to: UInt32.self, capacity: Self.topK)
+        let predicted = pilot.predictedIndices.contents()
+            .bindMemory(to: UInt32.self, capacity: Self.topK)
+        let realIds = (0..<Self.topK).map { real[$0] }
+        let predictedIds = (0..<Self.topK).map { predicted[$0] }
+        #expect(realIds == predictedIds)
+        #expect(Set(realIds).count == Self.topK,
+                "top-k collapsed — the fixture stopped exercising the ranking")
+        #expect(realIds.allSatisfy { $0 < UInt32(Self.numRouted) })
+    }
+}
diff --git a/Tests/Mference/Core/Runtime/InklingGenerationRegressionTests.swift b/Tests/Mference/Core/Runtime/InklingGenerationRegressionTests.swift
index 5e4e4b9..a09ee62 100644
--- a/Tests/Mference/Core/Runtime/InklingGenerationRegressionTests.swift
+++ b/Tests/Mference/Core/Runtime/InklingGenerationRegressionTests.swift
@@ -98,6 +98,45 @@ import Metal
                 "completion is implausibly short:\n\(text)")
     }
 
+    /// Shadow speculative prefetch must be a pure transport change: greedy
+    /// tokens with the pilot+shadow path on are byte-identical to `off`.
+    /// Env-gated on the real install like the suite's other tests.
+    @Test func shadowPrefetchKeepsGreedyTokensIdentical() async throws {
+        guard let path = ProcessInfo.processInfo
+            .environment["MFERENCE_INKLING_GTURBO"] else { return }
+        let ctx = try MetalContext()
+        let cfg = try #require(ArchConfig.knownArchitectures[.inklingSmall])
+        // Eight slots force eviction churn, so speculation actually issues.
+        let model = try Model.load(directoryURL: URL(fileURLWithPath: path),
+                                   device: ctx.device,
+                                   expecting: cfg,
+                                   streamingMode: .pread(slotCount: 8))
+        let runner = try RealForwardRunner(model: model, context: ctx,
+                                           maxContext: 64)
+        let logits = ctx.device.makeBuffer(
+            length: cfg.vocabSize * MemoryLayout<Float16>.stride,
+            options: .storageModeShared)!
+
+        func decode(steps: Int) async throws -> [UInt32] {
+            var out: [UInt32] = []
+            var token: Int32 = 200_028
+            for position in 0..<steps {
+                try await runner.produce(token: token, position: position,
+                                         into: logits)
+                out.append(runner.lastGreedyToken)
+                token = Int32(runner.lastGreedyToken)
+            }
+            return out
+        }
+
+        runner.speculativePrefetchMode = .off
+        let reference = try await decode(steps: 8)
+        runner.reset()
+        runner.speculativePrefetchMode = .shadow
+        let shadowed = try await decode(steps: 8)
+        #expect(reference == shadowed)
+    }
+
     /// The head guard itself: with a finite residual stream the epilogue must
     /// report zero non-finite logits, so `produce()` cannot throw
     /// `InklingHeadError`. Cheap — one token, no generation loop.

From 47526f200d21accdfc71c2c3c23ff83e1b1f4a52 Mon Sep 17 00:00:00 2001
From: NeelM0906 <neelanjan.mitra@acti.ai>
Date: Thu, 20 Aug 2026 16:13:14 -0400
Subject: [PATCH 3/6] perf: pipeline Inkling prefill expert streaming at depth
 1

prefillInklingChunk's routed loop serialized fetch->encode->drain, so
the SSD never overlapped GPU work (the deficiency documented in
docs/INKLING_SMALL.md). Now expert e+1 is planned and preaded while
expert e's GLU runs, with misses placed only in slots the in-flight
command buffer does not touch; accumulation order is unchanged (serial
queue + hazard tracking), so output stays byte-identical. Uses the same
PrefillRoutedTileScheduler contract as the other families.
MFERENCE_INKLING_PREFILL_PIPELINE=0 is the kill-switch.
---
 .../Runtime/Inference/RealForwardRunner.swift | 81 ++++++++++++++++---
 1 file changed, 70 insertions(+), 11 deletions(-)

diff --git a/Sources/Mference/Runtime/Inference/RealForwardRunner.swift b/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
index 17122c4..347293b 100644
--- a/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
+++ b/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
@@ -467,6 +467,12 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
     /// byte-identical). `MFERENCE_SLOT_MAP=0` is the kill-switch.
     static let slotMapEnabledDefault =
         ProcessInfo.processInfo.environment["MFERENCE_SLOT_MAP"] != "0"
+    /// Inkling prefill expert streaming: depth-1 pipeline — pread expert e+1
+    /// while expert e's GLU runs, misses placed only in slots the in-flight
+    /// command buffer does not touch. `MFERENCE_INKLING_PREFILL_PIPELINE=0`
+    /// is the kill-switch back to the serialized fetch->encode->drain loop.
+    static let inklingPrefillPipelineEnabled =
+        ProcessInfo.processInfo.environment["MFERENCE_INKLING_PREFILL_PIPELINE"] != "0"
     var slotMapEnabled = RealForwardRunner.slotMapEnabledDefault
     /// Debug discriminator: encode the whole slot-map chain but feed the
     /// lookup an all-empty table, so the guarded kernels always no-op and
@@ -4473,10 +4479,47 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
             func now() -> UInt64 {
                 breakdown ? clock_gettime_nsec_np(CLOCK_UPTIME_RAW) : 0
             }
-            for range in routedRanges {
+            // Depth-1 pipeline: while the GPU runs expert e's GLU, the CPU
+            // plans and preads expert e+1 into slots the in-flight buffer
+            // does not touch (`avoidingSlots`), so the fetch overlaps GPU
+            // work. `acc`/`act` hazards need no explicit sync: consecutive
+            // buffers on the serial queue are hazard-tracked on those
+            // resources, so expert accumulation order — and therefore the
+            // output — is byte-identical to the serialized loop. When slot
+            // pressure leaves no avoiding plan (or the pipeline is switched
+            // off), the next iteration drains first and demand-fetches,
+            // which *is* the old serialized behavior.
+            let tileScheduler = PrefillRoutedTileScheduler(
+                config: Self.prefillRoutedTileSchedulerConfig)
+            var pendingExpertCB: (cb: MTLCommandBuffer, slots: [Int])?
+            var prefetched: (expert: Int, plan: RoutedExpertFetchPlan,
+                             blob: TensorView)?
+            func drainPendingExpertCB() throws {
+                guard let pending = pendingExpertCB else { return }
+                pendingExpertCB = nil
+                let tDrain = now()
+                waitForCompletion(pending.cb)
+                if breakdown { Self.prefillDrainNanos &+= now() - tDrain }
+                if let error = pending.cb.error { throw error }
+            }
+            for (index, range) in routedRanges.enumerated() {
                 let t0 = now()
-                let blob = try await model.fetchRoutedExperts(layer: L,
-                                                              experts: [range.expert])[0]
+                let blob: TensorView
+                let blobSlots: [Int]
+                if let ready = prefetched, ready.expert == range.expert {
+                    blob = ready.blob
+                    blobSlots = ready.plan.assignedSlots
+                    prefetched = nil
+                } else {
+                    prefetched = nil
+                    try drainPendingExpertCB()
+                    guard let plan = try model.planRoutedExperts(
+                        layer: L, experts: [range.expert]) else {
+                        throw ModelError.routedExpertPlanUnavailable(layer: L)
+                    }
+                    blob = try await model.fetchRoutedExperts(plan: plan)[0]
+                    blobSlots = plan.assignedSlots
+                }
                 let t1 = now()
                 let cb = ctx.queue.makeCommandBuffer()!
                 prefillGLU.encode(commandBuffer: cb,
@@ -4490,20 +4533,36 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
                                   params: expertParams(range, base: Int(blob.offset)))
                 cb.commit()
                 let t2 = now()
-                // The next fetch may evict this expert's slot, and the shared
-                // `act` tile is reused, so drain before moving on. This
-                // serialization is why fetch does not overlap GPU work; see
-                // PrefillRoutedTileScheduler for the pipelined alternative the
-                // other families use.
-                waitForCompletion(cb)
                 if breakdown {
-                    let t3 = clock_gettime_nsec_np(CLOCK_UPTIME_RAW)
                     Self.prefillFetchNanos &+= t1 - t0
                     Self.prefillEncodeNanos &+= t2 - t1
-                    Self.prefillDrainNanos &+= t3 - t2
                     Self.prefillExpertCount &+= 1
                 }
+                try drainPendingExpertCB()
+                pendingExpertCB = (cb, blobSlots)
+
+                if Self.inklingPrefillPipelineEnabled,
+                   index + 1 < routedRanges.count {
+                    let nextExpert = routedRanges[index + 1].expert
+                    let nextPlan = try model.planRoutedExpertsIfPossible(
+                        layer: L,
+                        experts: [nextExpert],
+                        avoidingSlots: Set(blobSlots))
+                    let decision = tileScheduler.decide(
+                        PrefillRoutedTileSchedulerInput(
+                            hasPendingTile: true,
+                            pendingAssignedSlots: blobSlots,
+                            avoidingSlotPlanAvailable: nextPlan != nil))
+                    if case .prefetchNext = decision, let nextPlan {
+                        let tFetch = now()
+                        let nextBlob =
+                            try await model.fetchRoutedExperts(plan: nextPlan)[0]
+                        if breakdown { Self.prefillFetchNanos &+= now() - tFetch }
+                        prefetched = (nextExpert, nextPlan, nextBlob)
+                    }
+                }
             }
+            try drainPendingExpertCB()
             totalIoNanos &+= clock_gettime_nsec_np(CLOCK_UPTIME_RAW) - tIoStart
 
             // Tail: one causal channel-wise dispatch replaces N narrowing,

From 5b1633393602e8082571fda5c3f852871c9eb0dd Mon Sep 17 00:00:00 2001
From: NeelM0906 <neelanjan.mitra@acti.ai>
Date: Thu, 20 Aug 2026 16:13:49 -0400
Subject: [PATCH 4/6] docs: record Inkling streaming pipeline + prefetch port
 status

---
 docs/INKLING_SMALL.md | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/docs/INKLING_SMALL.md b/docs/INKLING_SMALL.md
index d6f8b9e..ad30bee 100644
--- a/docs/INKLING_SMALL.md
+++ b/docs/INKLING_SMALL.md
@@ -465,9 +465,16 @@ Those measurements identified two follow-ups:
 1. **Batched prefill attention was the main compute opportunity.** The
    chunk-wide portable and Apple10 TensorOps paths described below now cover
    Inkling's relative-position attention, including a wrapped sliding KV ring.
-2. **Expert streaming is serialized.** `prefillInklingChunk` awaits a fetch,
-   encodes, then drains, so the fetch never overlaps GPU work.
-   `PrefillRoutedTileScheduler` implements the pipelined alternative.
+2. **Expert streaming is serialized.** `prefillInklingChunk` awaited a fetch,
+   encoded, then drained, so the fetch never overlapped GPU work.
+   *Addressed 2026-08-20:* the loop now runs a depth-1 pipeline on the
+   `PrefillRoutedTileScheduler` contract — expert e+1 is planned and preaded
+   while expert e's GLU runs, misses placed only in slots the in-flight
+   buffer does not touch, output byte-identical.
+   `MFERENCE_INKLING_PREFILL_PIPELINE=0` restores the serialized loop.
+   Decode-side, the DSV4 pilot/shadow speculative prefetch is also ported
+   (`SpeculativeRouterInkling`, opt-in via `MFERENCE_SPEC_PREFETCH=shadow`
+   until an install A/B accepts a default).
 
 Method note, recorded because it cost real time: the first three attempts at
 this ranked the levers from dispatch counts and a per-pair cost fitted across

From e5be01b900df7f85385026f0fd46d9caed891de0 Mon Sep 17 00:00:00 2001
From: NeelM0906 <neelanjan.mitra@acti.ai>
Date: Thu, 20 Aug 2026 16:16:19 -0400
Subject: [PATCH 5/6] =?UTF-8?q?docs:=20S4=20encode-ahead=20hazard=20analys?=
 =?UTF-8?q?is=20=E2=80=94=20queue-stall=20deadlock=20made=20precise?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../2026-08-20-s4-encode-ahead-hazard.md      | 61 +++++++++++++++++++
 1 file changed, 61 insertions(+)
 create mode 100644 docs/superpowers/specs/2026-08-20-s4-encode-ahead-hazard.md

diff --git a/docs/superpowers/specs/2026-08-20-s4-encode-ahead-hazard.md b/docs/superpowers/specs/2026-08-20-s4-encode-ahead-hazard.md
new file mode 100644
index 0000000..be2422f
--- /dev/null
+++ b/docs/superpowers/specs/2026-08-20-s4-encode-ahead-hazard.md
@@ -0,0 +1,61 @@
+# S4 encode-ahead: the ordering hazard, made precise
+
+Status: analysis only — no implementation. Written so the next attempt does
+not rediscover the constraint the hard way.
+
+## What S4 wants
+
+`15-gpu-slot-map.md` measured that the slot-map gain lands below projection
+because "the router-wake event wait itself remains on every layer." At 96
+slots 50.7% of layer-steps are all-hit, yet the CPU still serializes every
+layer: encode L → commit → wait wake L → encode L+1. S4 = commit layer L+1's
+cb1 *before* layer L's wake, so consecutive all-hit layers flow on the GPU
+with no CPU-paced gap.
+
+## The hazard, precisely
+
+The natural design — pre-encode L+1's cb1 and gate its start on a shared
+event that is signaled once layer L's routed contribution is in `hidden` —
+deadlocks on the miss path:
+
+1. Metal executes command buffers on one queue in commit order, and an
+   `encodeWaitForEvent` stalls the *queue*, not just the buffer.
+2. On a miss layer, the routed command buffer is encoded by the CPU after
+   the wake and committed *after* the pre-committed L+1 cb1.
+3. L+1 cb1 stalls on the event; the routed CB behind it can never start;
+   the event it waits for can never be signaled. Deadlock.
+
+Conditional GPU-side signaling cannot rescue this: `encodeSignalEvent` is
+unconditional in the command stream, so the guarded FFN buffer cannot
+signal "only if all-hit."
+
+## Viable shapes (all unbuilt)
+
+- **Second queue for the speculative chain.** The routed fallback stays on
+  the primary queue; pre-encoded L+1 work waits on its own queue without
+  blocking the fallback. Caution: the plain second-queue experiment
+  (ORCH-10) *lost* throughput; this variant is different (event-gated, not
+  free-running) but must re-run that A/B.
+- **Guarded mega-buffer (Task 11 convergence).** Encode the whole token
+  step in one CB with every layer's routed FFN guarded by its slot-lookup
+  flag, and a *re-encode* fallback pass for the miss layers after the CB
+  completes. Miss layers then cost a second partial pass — only pays while
+  the all-hit rate stays high; needs the arg-buffer-reuse caution from the
+  ORCH table (prior reuse experiment lost 9% on prefill).
+- **CPU-signaled event, wake off the GPU path.** Keep the CPU wake, but
+  pre-encode+commit L+1 gated on a CPU-signaled event value; at an all-hit
+  wake the CPU's only work is `event.signaledValue = v` (microseconds)
+  instead of encode+commit (~hundreds of µs × 40 layers). Same queue-stall
+  constraint applies to the miss path, so this still needs one of the two
+  shapes above for the fallback, or a bounded speculation depth of exactly
+  one layer with the routed fallback encoded into the *same* pre-committed
+  buffer as predicated no-op work.
+
+## Measurement gate
+
+Any attempt must show, on the community protocol with the real Qwen 3.6
+install: byte-identical outputs, and a decode gain consistent with the
+attribution ceiling (~68 tok/s gap-free bound from
+`13-dsv4-streaming-iterations.md` methodology applied to Qwen). Toy-model
+parity (QwenSlotMapParityTests pattern) gates correctness before any real
+bench.

From 91da507f038721d05c51965164f515d5358f81e1 Mon Sep 17 00:00:00 2001
From: NeelM0906 <neelanjan.mitra@acti.ai>
Date: Thu, 20 Aug 2026 21:44:31 -0400
Subject: [PATCH 6/6] fix: drain the in-flight prefill expert buffer before
 propagating errors

Codex review on #21: a throw mid-pipeline (failed prefetch pread, drain
surfacing a GPU error) exited prefillInklingChunk with the pending
command buffer still running against the persistent act/acc scratch,
racing an immediate reset or retry. A scoped drain now waits it out on
every exit path.
---
 .../Runtime/Inference/RealForwardRunner.swift         | 11 +++++++++++
 1 file changed, 11 insertions(+)

diff --git a/Sources/Mference/Runtime/Inference/RealForwardRunner.swift b/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
index 347293b..a7398aa 100644
--- a/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
+++ b/Sources/Mference/Runtime/Inference/RealForwardRunner.swift
@@ -4494,6 +4494,17 @@ public final class RealForwardRunner: ChunkedPrefillRunner, ContextWindowReporti
             var pendingExpertCB: (cb: MTLCommandBuffer, slots: [Int])?
             var prefetched: (expert: Int, plan: RoutedExpertFetchPlan,
                              blob: TensorView)?
+            // Error-path backstop: any throw below (a failed prefetch pread,
+            // a drain surfacing a GPU error) must not leave the in-flight
+            // buffer running against the persistent act/acc scratch — an
+            // immediate reset or retry would race it. The success path
+            // drains and clears `pendingExpertCB`, so this no-ops there.
+            defer {
+                if let pending = pendingExpertCB {
+                    waitForCompletion(pending.cb)
+                    pendingExpertCB = nil
+                }
+            }
             func drainPendingExpertCB() throws {
                 guard let pending = pendingExpertCB else { return }
                 pendingExpertCB = nil