Several conversations at once: batch slots, pipelined layer split, per-stage dense weights

The engine's --serve can now keep up to N conversations open (--batch N) and decode them together, one token
each per verify window; with a layer split, --batch-groups G pipelines the slot groups through the GPUs, and
--trim-stage-weights makes every GPU load only its own layers' dense weights (VRAM back to the expert cache).
The server runs concurrent requests in the slots. See docs/BATCHING.md.

docs/BATCHING.md

@@ -0,0 +1,94 @@
+# Several conversations at once (batch slots)
+
+By default Strata serves **one request at a time**: the others wait in the server's queue. With `--batch N`
+the engine keeps up to N conversations open and decodes them **together**: every verify window then carries one
+token of each conversation, so the dense weights, the shared expert and the head are read once per window for
+all of them. Combined with a layer split across several GPUs and `--batch-groups`, the cards also work on
+different conversations at the same time instead of waiting for each other.
+
+It is opt-in and changes nothing when the options are absent.
+
+## Turning it on
+
+Add the options to the `args` list of the model's config (`strata-<model>.json`) and restart:
+
+```
+"args": [ ..., "--batch", "8", "--batch-groups", "4", "--trim-stage-weights" ],
+"layer_split": "12,24,36"
+```
+
+| Option | What it does |
+| --- | --- |
+| `--batch N` (2..8) | up to N conversations decoded together. Each slot gets its own state (a session carved like the stage's own: GDN recurrence, QSA K/V and indexer, PLE history) on every GPU of the split. |
+| `--batch-groups G` | with a layer split: the N slots in G groups that flow through the GPUs as a pipeline (GPU k runs one group while GPU k+1 runs another). G must divide N. 1 = all slots in one window, GPU after GPU. |
+| `--trim-stage-weights` | with an **explicit** `--layer-split` (e.g. `12,24,36`, not `auto`): every GPU loads only the dense weights of its own layers instead of the whole model's. The VRAM this frees goes to the expert cache - which is what makes batching pay, since several conversations together touch more distinct experts than one. Useful without `--batch` too. |
+
+The server needs nothing else: it reads `--batch` / `--batch-groups` from the engine's arguments.
+
+## How the server uses the slots
+
+- **One request alone** runs on the usual solo path (verify windows with MTP drafts): the fastest single stream.
+- **When a second request arrives**, the first is stopped (`STOP`) and continues in a batch slot with its prompt
+  plus what it generated so far - the engine's prompt cache holds exactly that, so nothing is read again - and the
+  new request is admitted next to it.
+- **Each admission** reads the request's prompt through the usual prompt path (prompt cache and conversation
+  checkpoints included) and produces its first token there; the state is then copied into the slot. Admissions
+  are taken one at a time; the slots already decoding pause meanwhile.
+- Slots are assigned so that consecutive requests land in different pipeline groups.
+- A client that disconnects stops its slot (`BSTOP`); the others go on.
+
+## Exactness
+
+A batch row's arithmetic is the single-token window's, so with greedy decoding **every conversation of a batch
+produces exactly the tokens it produces alone** - verified token by token for 8 concurrent conversations of 150
+tokens, with and without the pipeline (`tools/batch_test.py`). Two settings make that comparison exact:
+
+- `STRATA_IQ_MT_MIN=1` (the multi-token CPU expert kernels for every group, as for the solo path's own
+  exactness tests), and
+- `--pcie-frac 0`: the PCIe share of the missed experts is chosen per window from the window's misses, so the
+  same expert can run on the GPU in one window and on the CPU in another, which rounds differently. With a share,
+  outputs stay coherent but can drift apart after some tokens, as they would between two solo runs whose windows
+  differ.
+
+Sampled requests (temperature, top_p, top_k, min_p, seed) are drawn row by row with the solo window's
+counter-based draw (Philox(seed, position)).
+
+## Limits (for now)
+
+- Batch windows carry no MTP drafts: a conversation in a slot decodes one token per window (the solo path keeps
+  its drafts, which is why a request alone is not put in a slot).
+- Repetition / frequency / presence penalties are not applied in batch windows.
+- `--batch-groups` needs every stage on its own GPU.
+- The slot sessions take VRAM (each one like the stage's own session; a few hundred MB per slot at 64K context
+  with KV streaming) and pinned RAM for their streamed K/V.
+
+## Measured
+
+A 4-GPU layer split (4 x 16 GB, PCIe Gen3), IQ3_S, `--batch 8 --batch-groups 4 --trim-stage-weights`, through
+the HTTP server, 400 tokens per answer, temperature 0.7:
+
+| Concurrent requests | Per request | Total |
+| ---: | ---: | ---: |
+| 1 | 123 tok/s (solo path) | 120 tok/s |
+| 2 | 57 tok/s | 113 tok/s |
+| 4 | 51 tok/s | 205 tok/s |
+| 8 | 45 tok/s | 360 tok/s |
+
+`--trim-stage-weights` alone raised the share of experts held in VRAM on that machine from 76-85 % to 84-100 %
+per card.
+
+## Engine protocol (`--serve`)
+
+On top of `GEN` / `GENI`:
+
+| Line | Direction | Meaning |
+| --- | --- | --- |
+| `BGEN <slot> <max_new> [keys] <ids>` | in | read the prompt (as `GEN 1`), then continue in `<slot>` |
+| `BGENI <slot> <max_new> [keys] <file> <ids>` | in | the same with images |
+| `BADM <slot> <1/0>` | out | after the admission's `DONE`: 1 = it continues in the slot, 0 = it ended |
+| `BT <slot> <id>` | out | a token of that slot |
+| `BDONE <slot> <generated> <stop/length/cancel> <ms>` | out | the slot is free again |
+| `BSTOP <slot>` | in | end that slot at its next window |
+
+`tools/batch_test.py` drives the engine directly: the same prompts alone, then together, compared token by token,
+and the aggregate rate.

docs/MULTI_GPU.md

@@ -140,3 +140,8 @@ The Coder on an RTX 5080 + RTX 3090 (Ryzen 9 9950X3D), 32K context; details in
 - Leave out a much slower card when two already hold the model. An RTX 2080 Ti as a third card made the 5080 +
   3090 pair slower (68 / 90 tok/s decode): every extra card costs its own round per window.
 - More cards pay off when the model's routed experts do not fit the faster ones.
+
+## Several conversations at once
+
+With a layer split, `--batch N --batch-groups G --trim-stage-weights` decodes several conversations together and
+pipelines them through the cards: see [BATCHING.md](BATCHING.md).

include/strata/core/native_dense.hpp

@@ -19,8 +19,10 @@ public:
     ~NativeDense();
     NativeDense(const NativeDense&) = delete;
     NativeDense& operator=(const NativeDense&) = delete;
+    /// With layer_hi >= 0, only the `blk.<l>.` matrices with layer_lo <= l < layer_hi are uploaded (a layer
+    /// split's stage holds its own layers' projections, not the whole model's); the others keep data == nullptr.
     bool load(const std::vector<std::string>& shards, WeightTable& table, std::string& err,
-              bool include_ple_key = false);
+              bool include_ple_key = false, int64_t layer_lo = 0, int64_t layer_hi = -1);
     /// Plan v0.3 P1: the canonical tensor names `load` would serve natively from these shards (eligible name,
     /// supported type, 2-D), read from the GGUF headers only - so the canonical arena can skip them.
     static bool served_names(const std::vector<std::string>& shards, bool include_ple_key,

include/strata/core/verify.hpp

@@ -31,6 +31,8 @@
 #include <cuda_runtime.h>
 
 #include <atomic>
+#include <chrono>
+#include <map>
 #include <cstdint>
 #include <string>
 #include <vector>
@@ -119,6 +121,46 @@ public:
 
     /// Keep the first `n_keep` (1..T) tokens of the last window; advances `ss.ple_prev` by them.
     bool commit(int n_keep, std::string& err);
+
+    // ================================ SEVERAL SEQUENCES IN ONE WINDOW ================================
+    //
+    // A batch window holds S INDEPENDENT sequences, one token each: row s is slot s, at slot s's own position,
+    // reading and writing slot s's own state (GDN recurrence and conv history, QSA K/V and indexer, PLE history),
+    // which lives in `slots[s]` - a session carved like this verifier's own (same layer range, same max_cells).
+    // Everything that is per row already (hyper-connections, dense projections, router, shared expert, the
+    // routed experts on the GPU and the CPU, the head) runs once over the S rows, so the weights are read once
+    // per window for all the sequences.  Row s's arithmetic is the single-token window's, so a slot's greedy
+    // tokens are its solo greedy tokens (modulo the multi-token CPU kernel choice: STRATA_IQ_MT_MIN=1).
+    //
+    // Greedy only, no drafts (MTP) in a batch window.  `init_slots` once after `init` (S <= max_t); a layer
+    // split's stages each get their own sessions, and run_slots/commit_slots continue into the next stage.
+    bool init_slots(const std::vector<SessionState*>& slots, std::string& err);
+    int n_slots() const { return (int) slots_.size(); }
+    /// One batch window over slots [0, S): tokens[s] at positions pos[s]; out[s] = the greedy pick after it.
+    bool run_slots(int S, const int32_t* tokens, const int64_t* pos, PoolMultiFn pool, void* user, int32_t* out,
+                   std::string& err);
+    /// Keep every row of the last batch window: each slot's state advances by its one token.
+    bool commit_slots(std::string& err);
+
+    // ---- The stages of a layer split as a PIPELINE.  A batch window over the slot GROUP
+    // [base, base + S) is launched on ONE stage with its commit right behind it on the stage's stream (a batch window
+    // keeps every row, so the commit needs no host decision), and the host serves the rings of every stage that has
+    // a window in flight from one thread (batch_poll does not block).  Stage k can then run group g while stage k+1
+    // runs group g-1.  Rows of group `base` use hand-off rows [base, base + S), so groups never share a hand-off row.
+    bool batch_launch(int base, int S, const int32_t* tokens, const int64_t* pos, std::string& err);
+    /// 1 = this stage's window and commit are done (the last stage's picks are in batch_out), 0 = still running,
+    /// -1 = an error (err).  Serves every layer that has rung so far.
+    int batch_poll(PoolMultiFn pool, void* user, std::string& err);
+    bool batch_busy() const { return b_running_; }
+    /// A slot's sampling (temperature / top_p / top_k / min_p / seed; penalties are not applied in batch windows):
+    /// its row is drawn again on the last stage with Philox(seed, position), as a solo window draws it.  Greedy by
+    /// default.  Set on the first stage, it reaches the last.
+    void set_slot_sampling(int slot, const strata::kernels::SamplerParams& sp) {
+        if (slot >= 0 && slot < (int) slot_sp_.size()) slot_sp_[(size_t) slot] = sp;
+        if (next_) next_->set_slot_sampling(slot, sp);
+    }
+    const int32_t* batch_out() const { return b_out_; }
+    bool last_stage() const { return g_ != nullptr && le_ == g_->n_layers; }
     /// commit() returns without waiting for its graph (a single-GPU session sets it): the next window follows it on
     /// the same stream and the drafter reads nothing it writes, so it overlaps the draft. Whoever reads or writes
     /// the session from another stream or the host afterwards (a new request, a checkpoint, a snapshot, the prompt
@@ -162,6 +204,29 @@ public:
 
 private:
     bool capture(int T, std::string& err);
+    // batch windows (see init_slots)
+    std::vector<SessionState*> slots_;
+    bool batch_rec_ = false;               ///< record_window is capturing a batch window
+    int row_base_ = 0;                     ///< ... over slots [row_base_, row_base_ + T)
+    bool last_batch_ = false;              ///< the last run was a batch window (set_plan_slot: one group)
+    std::map<int, cudaGraphExec_t> exec_bm_, commit_bm_;   ///< key base * 16 + S
+    int last_base_ = 0;
+    // batch_launch / batch_poll
+    bool b_running_ = false;
+    int64_t b_k_ = 0, b_steps_ = 0;
+    std::chrono::steady_clock::time_point b_last_;
+    int32_t b_out_[8] = {};
+    std::vector<strata::kernels::SamplerParams> slot_sp_;
+    bool sample_rows(int base, int S, std::string& err);   ///< the sampled slots' rows of the last batch window
+    int32_t* h_commitb_ = nullptr; int32_t* m_commitb_ = nullptr;   // per slot [1, 0, pos, -1 ..], stride 2 + max_t
+    int32_t* commitb_ = nullptr;
+    float* tail_snap_b_ = nullptr;         ///< per (slot, QSA layer) indexer tail snapshot
+    void* arena_b_ = nullptr;
+    int64_t last_pos_b_[8] = {};
+    bool capture_batch(int base, int S, std::string& err);
+    void collect_profile();   ///< STRATA_VERIFY_PROFILE: add the last window's stamps to prof_sum_
+    bool capture_commit_batch(int base, int S, std::string& err);
+    bool stage_batch(int base, int S, const int32_t* tokens, const int64_t* pos, std::string& err);
     strata::kernels::SamplerParams sampling_ = [] {
         strata::kernels::SamplerParams s;
         s.greedy = true;

serve/server.py

@@ -21,6 +21,7 @@ talks to it over stdin/stdout; `MockEngine` is a scripted stand-in that makes ev
 from __future__ import annotations
 
 import argparse
+import contextlib
 import collections
 import base64
 import hashlib
@@ -292,6 +293,9 @@ def narrate_start(log_path: str, offset: int, args: list, done: threading.Event,
             print(f"[strata] still starting ({time.time() - t0:.0f} s) - please wait ...", flush=True)
 
 
+EOS_IDS = {248044, 248046}   # <|endoftext|>, <|im_end|>: the engine's default --eos-ids
+
+
 class StrataEngine:
     """The resident engine: `strata --serve` reads `GEN <max_new> <ids>` lines and streams `T <id>` lines, then
     `DONE ...`.  Requests are serialized by the service's FIFO, so one pipe is enough.
@@ -368,16 +372,38 @@ class StrataEngine:
         self.ended = False                              # READY: alive from here (restart() set it True, #344)
         # the engine's stdout on a thread, so a request can wait with a timeout (heartbeats, cancel checks)
         self.lines: queue.Queue = queue.Queue()
+        # --batch: concurrent requests in the engine's batch slots (see generate_batched)
+        self.batch = int(args[args.index("--batch") + 1]) if "--batch" in args else 0
+        groups = int(args[args.index("--batch-groups") + 1]) if "--batch-groups" in args else 1
+        groups = groups if self.batch and groups > 0 and self.batch % groups == 0 else 1
+        gs = self.batch // groups if self.batch else 0
+        # slots in the order that spreads requests over the pipeline's groups first: 0, gs, 2gs, .., 1, gs+1, ..
+        self.slot_order = [g * gs + t for t in range(gs) for g in range(groups)]
+        self.slot_q = [queue.Queue() for _ in range(self.batch)]
+        self.slot_busy = [False] * self.batch
+        self.slot_cv = threading.Condition()
+        self.waiting = 0                                # requests waiting for the control lines (ctl)
+        self.ctl = threading.Lock()                     # one admission or solo request on the control lines at a time
+        self.wlock = threading.Lock()                   # stdin writes from several request threads
         self.pump = threading.Thread(target=self._pump, daemon=True)
         self.pump.start()
 
     def _pump(self):
         proc, lines = self.proc, self.lines             # this process's: a restart replaces both (#344)
+        slot_q = self.slot_q
         for line in proc.stdout:
+            if line.startswith(("BT ", "BDONE ")) and slot_q:   # --batch: a batch slot's own lines
+                try:
+                    slot_q[int(line.split()[1])].put(line)
+                    continue
+                except (IndexError, ValueError):
+                    pass
             lines.put(line)
         if self.proc is proc:                           # a killed engine's pump must not mark its successor dead
             self.ended = True                           # its output closed: it is gone, even before the OS says so
         lines.put(None)
+        for q in slot_q:
+            q.put(None)
 
     def death_note(self) -> str:
         """Why the engine most likely ended, from the end of its log: its own watchdog (issue #29), else RAM."""
@@ -498,10 +524,171 @@ class StrataEngine:
         on = sampling.get("experimental_speed_projection")
         return f" cvec={int(on)}" if isinstance(on, bool) else ""
 
+    def _send(self, text: str):
+        try:
+            with self.wlock:
+                self.proc.stdin.write(text + "\n")
+                self.proc.stdin.flush()
+        except OSError:                                  # the pipe is gone: the engine died (not the client)
+            raise EngineDied(f"the engine stopped unexpectedly (exit code {self.exit_code()})") from None
+
+    def _control(self, cancel, on_token, stop_when=None):
+        """Reads the control lines of the request on them (GEN / BGEN), yielding None heartbeats.  Calls on_token(id)
+        for each `T`; returns ("done", None) at DONE, or ("badm", continues) at BADM (after DONE).  `stop_when()`
+        true sends STOP once (the request is then read to its DONE)."""
+        stopped = False
+        while True:
+            try:
+                line = self.lines.get(timeout=10.0)
+            except queue.Empty:
+                if cancel.is_set() and not stopped:
+                    self._send("STOP")
+                    stopped = True
+                yield None
+                continue
+            if line is None:
+                raise EngineDied(f"the engine stopped unexpectedly (exit code {self.exit_code()})")
+            if line.startswith("T "):
+                on_token(int(line[2:]))
+                if not stopped and (cancel.is_set() or (stop_when is not None and stop_when())):
+                    self._send("STOP")
+                    stopped = True
+                yield False                              # a token is pending (not a heartbeat)
+            elif line.startswith("PP "):
+                f = line.split()
+                if len(f) >= 3 and f[1].isdigit() and f[2].isdigit():
+                    self.progress = (int(f[1]), int(f[2]))
+                    self.prefill_tok_s_mean = float(f[4]) if len(f) >= 5 else None
+                yield None
+            elif line.startswith("DONE"):
+                self._parse_done(line)
+                self._last_done = line
+            elif line.startswith("BADM "):
+                f = line.split()
+                self._ctl_result = ("badm", len(f) >= 3 and f[2] == "1")
+                return
+            elif line.startswith("ERR"):
+                raise ValueError(line[4:].strip())
+            if line.startswith("DONE") and self._ctl_mode == "solo":
+                self._ctl_result = ("done", None)
+                return
+
+    def generate_batched(self, ids, max_new, sampling, cancel, embeddings=None):
+        """--batch.  Alone (no slot busy, nobody waiting): the solo path (GEN, with drafts: the fastest stream)
+        - and when another request arrives meanwhile, this one is STOPped and continues in a batch slot (BGEN with
+        its prompt + what it generated: the engine reuses that prefix).  Otherwise: BGEN into a free slot, then the
+        slot's own BT lines until BDONE.  Several requests run at once; the control lines (prompt reading, admission)
+        are taken one request at a time."""
+        self.progress = None
+        keys = self.sampling_keys(sampling or {})
+        out: list[int] = []
+        pending: list[int] = []
+        with self.slot_cv:
+            self.waiting += 1
+        try:
+            while not self.ctl.acquire(timeout=10.0):
+                if cancel.is_set():
+                    return
+                yield None
+        finally:
+            with self.slot_cv:
+                self.waiting -= 1
+        holding = True
+        slot = None
+        try:
+            with self.slot_cv:
+                alone = not any(self.slot_busy) and self.waiting == 0
+            prompt, left = list(ids), int(max_new)
+            if alone and left > 1:
+                head = f"GENI {left}{keys} {embeddings}" if embeddings else f"GEN {left}{keys}"
+                self._send(f"{head} {','.join(str(int(t)) for t in prompt)}")
+                self._ctl_mode, self._ctl_result = "solo", None
+                def others():
+                    with self.slot_cv:
+                        return self.waiting > 0
+                for x in self._control(cancel, pending.append, stop_when=others):
+                    while pending:
+                        t = pending.pop(0)
+                        out.append(t)
+                        yield t
+                    if x is None:                       # a heartbeat (False: a token, flushed above)
+                        yield None
+                while pending:
+                    t = pending.pop(0)
+                    out.append(t)
+                    yield t
+                finish = (self.last or {}).get("finish") if isinstance(self.last, dict) else None
+                left = int(max_new) - len(out)
+                if cancel.is_set() or left <= 0 or finish in ("stop", "length") or (out and out[-1] in EOS_IDS):
+                    return
+                prompt = list(ids) + out                # promoted: it continues in a batch slot from here
+                embeddings = embeddings                 # (an image prompt keeps its embeddings file)
+            # a free slot (they free themselves at BDONE, which needs no control lines)
+            with self.slot_cv:
+                while True:
+                    slot = next((b for b in self.slot_order if not self.slot_busy[b]), None)
+                    if slot is not None:
+                        self.slot_busy[slot] = True
+                        break
+                    self.slot_cv.wait(timeout=10.0)
+                    if cancel.is_set():
+                        return
+            while not self.slot_q[slot].empty():
+                self.slot_q[slot].get_nowait()
+            head = f"BGENI {slot} {left}{keys} {embeddings}" if embeddings else f"BGEN {slot} {left}{keys}"
+            self._send(f"{head} {','.join(str(int(t)) for t in prompt)}")
+            self._ctl_mode, self._ctl_result = "batch", None
+            for x in self._control(cancel, pending.append):
+                while pending:
+                    yield pending.pop(0)
+                if x is None:
+                    yield None
+            while pending:
+                yield pending.pop(0)
+            cont = bool(self._ctl_result and self._ctl_result[1])
+            self.ctl.release()
+            holding = False
+            if not cont:
+                return
+            stopped = False
+            while True:
+                try:
+                    line = self.slot_q[slot].get(timeout=10.0)
+                except queue.Empty:
+                    if cancel.is_set() and not stopped:
+                        self._send(f"BSTOP {slot}")
+                        stopped = True
+                    yield None
+                    continue
+                if line is None:
+                    raise EngineDied(f"the engine stopped unexpectedly (exit code {self.exit_code()})")
+                if line.startswith("BT "):
+                    if cancel.is_set():
+                        if not stopped:
+                            self._send(f"BSTOP {slot}")
+                            stopped = True
+                        continue
+                    yield int(line.split()[2])
+                elif line.startswith("BDONE "):
+                    f = line.split()
+                    if len(f) >= 5 and isinstance(self.last, dict):
+                        self.last = {**self.last, "finish": f[3], "decode_ms": float(f[4])}
+                    return
+        finally:
+            if holding:
+                self.ctl.release()
+            if slot is not None:
+                with self.slot_cv:
+                    self.slot_busy[slot] = False
+                    self.slot_cv.notify_all()
+
     def generate(self, ids, max_new, sampling, cancel, embeddings=None):
         """Yields token ids, and None as a heartbeat every 10 s while the engine is quiet (reading a long prompt):
         the HTTP layer turns it into an SSE comment, which keeps clients' watchdogs calm and notices a client that
         has gone.  A consumer that stops early (or `cancel`) makes the engine STOP, so it does not run to max_new."""
+        if getattr(self, "batch", 0):
+            yield from self.generate_batched(ids, max_new, sampling, cancel, embeddings)
+            return
         self.progress = None
         self.prefill_tok_s_mean = None
         # an image request takes the same sampling keys as text (#75: it used to decode greedily whatever was asked)
@@ -1447,7 +1634,8 @@ class Service:
         with self.status_lock:
             self.status["queued"] += 1
         try:
-            with self.fifo:
+            # --batch: the engine runs several requests at once (StrataEngine.generate_batched orders them)
+            with (contextlib.nullcontext() if getattr(self.engine, "batch", 0) else self.fifo):
                 try:
                     with self.status_lock:
                         if trace is not None:

src/core/native_dense.cpp

@@ -1,4 +1,5 @@
 #include "strata/core/native_dense.hpp"
+#include <cstdlib>
 #include "strata/core/weights.hpp"
 #include "strata/artifact/gguf_reader.hpp"
 #include "strata/kernels/native_mmvq.hpp"
@@ -69,7 +70,12 @@ NativeDense::~NativeDense() {
 }
 
 bool NativeDense::load(const std::vector<std::string>& shards, WeightTable& table, std::string& err,
-                       bool include_ple_key) {
+                       bool include_ple_key, int64_t layer_lo, int64_t layer_hi) {
+    auto outside = [&](const std::string& name) {   // a blk.<l>. tensor of another stage's layers
+        if (layer_hi < 0 || name.rfind("blk.", 0) != 0) return false;
+        const long l = std::strtol(name.c_str() + 4, nullptr, 10);
+        return l < layer_lo || l >= layer_hi;
+    };
     if (scratch_ || !weights_.empty()) { err = "native dense: already loaded"; return false; }
     if (shards.empty()) { err = "native dense: at least one GGUF shard is required"; return false; }
     try {
@@ -138,7 +144,7 @@ bool NativeDense::load(const std::vector<std::string>& shards, WeightTable& tabl
                 }
             }
             for (const auto& tensor : gguf.tensors()) {
-                if (!eligible(tensor, include_ple_key)) continue;
+                if (!eligible(tensor, include_ple_key) || outside(tensor.name)) continue;
                 if (!seen.insert(tensor.name).second) {
                     err = "native dense: duplicate tensor " + tensor.name; return false;
                 }

src/core/verify.cpp

@@ -176,6 +176,12 @@ Verifier::~Verifier() {
     for (auto& e : exec_)
         if (e) cudaGraphExecDestroy(e);
     if (commit_exec_) cudaGraphExecDestroy(commit_exec_);
+    for (auto& kv : exec_bm_)
+        if (kv.second) cudaGraphExecDestroy(kv.second);
+    for (auto& kv : commit_bm_)
+        if (kv.second) cudaGraphExecDestroy(kv.second);
+    if (arena_b_) cudaFree(arena_b_);
+    if (h_commitb_) cudaFreeHost(h_commitb_);
     if (cs_) cudaStreamDestroy(cs_);
     if (copy_) { cudaStreamSynchronize(copy_); cudaStreamDestroy(copy_); }
     if (commit_done_) cudaEventDestroy(commit_done_);
@@ -403,11 +409,15 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
     const int64_t TS = (s.idx_block - 1) * ID;
     const bool ple_on = ss.ple.ready() && ple_stage();
     auto Rt = [&](int t) { return R_ + (size_t) t * HC * N; };
-    const int G = (split_ && T >= 2) ? 2 : 1;
+    const int G = (split_ && T >= 2 && !batch_rec_) ? 2 : 1;   // a batch window is one group
     static const bool dec_batch = [] { const char* v = std::getenv("STRATA_DEC_BATCH"); return v == nullptr || std::atoi(v) != 0; }();
     auto stamp = [&](int64_t l, int i, int grp) { if (prof_on_ && grp == 0) gpu_stamp(prof_, (int) (l * kProfPer + i), cs); };
     const int tb_[2] = {0, (T + 1) / 2}, te_[2] = {G == 2 ? (T + 1) / 2 : T, T};
-    groups_[T] = G;
+    if (!batch_rec_) groups_[T] = G;
+    const int64_t nQall = g.n_qsa_layers();
+    // a batch window: row t is slot t, whose state lives in its own session
+    auto slot_ss = [&](int t) -> SessionState& { return batch_rec_ ? *slots_[(size_t) (row_base_ + t)] : ss; };
+    const int hrow0 = batch_rec_ ? row_base_ : 0;   // a slot group's own hand-off rows
 
     // ---- the window's inputs, from mapped staging
     copy_i32_from_mapped(tok_, m_tok_, T, cs);
@@ -423,9 +433,10 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
     const int64_t HB = Verifier::handoff_floats(g);
     if (lb_ > 0) {
         for (int t = 0; t < T; ++t) {
-            copy_from_mapped(Rt(t), hand_in_ + (size_t) t * HB, HC * N, cs);
-            copy_from_mapped(bo_ + (size_t) t * N, hand_in_ + (size_t) t * HB + HC * N, N, cs);
-            copy_from_mapped(inj2_ + (size_t) t * HC, hand_in_ + (size_t) t * HB + HC * N + N, HC, cs);
+            const size_t hr = (size_t) (hrow0 + t) * HB;
+            copy_from_mapped(Rt(t), hand_in_ + hr, HC * N, cs);
+            copy_from_mapped(bo_ + (size_t) t * N, hand_in_ + hr + HC * N, N, cs);
+            copy_from_mapped(inj2_ + (size_t) t * HC, hand_in_ + hr + HC * N + N, HC, cs);
         }
     } else if (const NativeEmbed* ne = native_embed()) {       // plan v0.3 P6: the GGUF-form table
         ne->gather_dev(tok_, T, emb_, cs);
@@ -480,14 +491,15 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
                 PleOut po;
                 po.normalized = normalized;
                 po.result = Rt(t);
+                float* hist = batch_rec_ ? slot_ss(t).ple_hist : ss.ple.hist;   // the row's own history
                 try {
-                    ple_block(ple_ + t * N, Rt(t), ss.ple.hist, ss.ple.w, po, ss.ple.scratch, cs);
-                    ple_history_advance(ss.ple.hist, normalized, cs);
+                    ple_block(ple_ + t * N, Rt(t), hist, ss.ple.w, po, ss.ple.scratch, cs);
+                    ple_history_advance(hist, normalized, cs);
                 } catch (const std::exception& e) {
                     err = std::string("verify PLE: ") + e.what();
                     return false;
                 }
-                copy_from_mapped(hist_snap_ + (size_t) t * HS, ss.ple.hist, HS, cs);
+                copy_from_mapped(hist_snap_ + (size_t) t * HS, hist, HS, cs);
             }
             pending = false;
         }
@@ -530,6 +542,15 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
                 native_quantize_q8_1(xm, xq_, (int) N, n, cs);
                 native_mmvq(wqkv->native_type, wqkv->native_data, xq_, qkv + (size_t) tb * C, (int) N, (int) C, n, cs);
                 stamp(l, 2, grp);
+                if (batch_rec_) {   // each row from its own slot's conv history, one row each
+                    for (int t = tb; t < te; ++t) {
+                        SessionState& sx = slot_ss(t);
+                        const float* cx = sx.gdn_state + (size_t) (gi - sx.gdn_ord0) * gdn_floats +
+                                          (uint64_t) g.ssm_state_size * g.ssm_v_heads * g.ssm_state_size;
+                        gdn_conv_l2_multi(cx, qkv + (size_t) t * C, (const float*) wc->data, hb + (size_t) t * C, (int) C,
+                                          (int) (2 * HK), EPS, 1, cs, 0);
+                    }
+                } else
                 gdn_conv_l2_multi(conv, qkv, (const float*) wc->data, hb, (int) C, (int) (2 * HK), EPS, n, cs, tb);
                 stamp(l, 3, grp);
                 gdn_ab_multi(xm, (const uint16_t*) wa->data, (const uint16_t*) wb->data, (const float*) wdt->data,
@@ -539,6 +560,15 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
                 native_mmvq(wg->native_type, wg->native_data, xq_, z_ + (size_t) tb * ZV, (int) N, (int) ZV, n, cs);
                 stamp(l, 5, grp);
                 // the recurrence from the untouched state over tokens [0, te); outputs only for this group's
+                if (batch_rec_) {   // each row's recurrence from its own slot's state, one token
+                    for (int t = tb; t < te; ++t) {
+                        SessionState& sx = slot_ss(t);
+                        float* stx = sx.gdn_state + (size_t) (gi - sx.gdn_ord0) * gdn_floats;
+                        gdn_step_norm_multi(stx, hb + (size_t) t * C, (int) C, gate + (size_t) t * HV, beta + (size_t) t * HV,
+                                            z_ + (size_t) t * ZV, (const float*) wnm->data, EPS, y_ + (size_t) t * ZV,
+                                            (int) HK, (int) HV, 1, nullptr, cs, 0);
+                    }
+                } else
                 gdn_step_norm_multi(state, hb, (int) C, gate, beta, z_, (const float*) wnm->data, EPS, y_, (int) HK,
                                     (int) HV, te, nullptr, cs, tb);
                 stamp(l, 6, grp);
@@ -583,8 +613,14 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
                     fwht256_inplace_cuda(vcur_ + tb * NKV * HD, (int64_t) n * NKV, cs);
                 }
                 stamp(l, 8, grp);
+                if (batch_rec_) {   // every slot's indexer tail, restored by its commit
+                    for (int t = tb; t < te; ++t)
+                        copy_from_mapped(tail_snap_b_ + ((size_t) (row_base_ + t) * nQall + qi) * TS, slot_ss(t).qsa_states[qi].idx_tail,
+                                         TS, cs);
+                } else
                 if (grp == 0) copy_from_mapped(tail_snap_ + (size_t) qi * TS, st.idx_tail, TS, cs);
                 for (int t = tb; t < te; ++t) {
+                    const QsaState& st = slot_ss(t).qsa_states[qi];   // the row's own K/V (ss's outside a batch)
                     const int32_t* step_t = step_ + t * kStepCount;
                     if (st.kv_hybrid) {   // K8V4: the unused half's lanes folded onto the used pool (layer.cpp)
                         kv_append_q8_step(st.k_q, st.k_q, st.k_scale, st.k_scale, st.page_table, step_t,
@@ -601,11 +637,13 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
                         kv_append_step(st.k_pool, st.v_pool, st.page_table, step_t, kcur_ + t * NKV * HD,
                                        vcur_ + t * NKV * HD, s, cs, &st.host);
                 }
-                const QsaIndexerBuffers ib{st.idx_tail, st.idx_dead, st.idx_pooled, st.idx_block_pos};
-                for (int t = tb; t < te; ++t)
+                for (int t = tb; t < te; ++t) {
+                    const QsaState& sx = slot_ss(t).qsa_states[qi];
+                    const QsaIndexerBuffers ib{sx.idx_tail, sx.idx_dead, sx.idx_pooled, sx.idx_block_pos};
                     native_qsa_indexer_append(idx_raw + t * ID, step_ + t * kStepCount + kStepPos, 0,
-                                              (const float*) wikn->data, EPS, ib, s, st.max_cells,
+                                              (const float*) wikn->data, EPS, ib, s, sx.max_cells,
                                               rope_scaling(), cs);
+                }
                 stamp(l, 9, grp);
                 native_mmvq(wq->native_type, wq->native_data, xq_, qfull_ + tb * NH * 2 * HD, (int) N, (int) (NH * 2 * HD),
                             n, cs);
@@ -638,6 +676,20 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
                 }
                 }
                 stamp(l, 10, grp);
+                if (batch_rec_) {   // each row selects and attends over its own slot's K/V
+                    for (int t = tb; t < te; ++t) {
+                        const QsaState& sx = slot_ss(t).qsa_states[qi];
+                        qsa_block_scores(sx.idx_pooled, sx.idx_dead, qidx_ + t * IQ * ID, step_ + t * kStepCount, 1,
+                                         max_blocks_, s, scores_ + (size_t) t * max_blocks_, cs);
+                        qsa_block_topk(scores_ + (size_t) t * max_blocks_, step_ + t * kStepCount, 1, max_blocks_, cap_, s,
+                                       sel_ + (size_t) t * cap_, cs);
+                        qsa_kv_resolve(sx, *g_, sel_ + (size_t) t * cap_, step_ + t * kStepCount, 1, cap_, cs);
+                        const QsaAttnPools px = qsa_attn_pools(sx);
+                        qsa_decode_attn_batch(qcur_ + t * NH * HD, px, sel_ + (size_t) t * cap_, step_ + t * kStepCount,
+                                              cap_, s, attn_scratch_ + (size_t) t * attn_scratch_floats_,
+                                              attn_ + t * NH * HD, 1, cs);
+                    }
+                } else {
                 qsa_block_scores(st.idx_pooled, st.idx_dead, qidx_ + tb * IQ * ID, step_ + tb * kStepCount, n, max_blocks_,
                                  s, scores_ + (size_t) tb * max_blocks_, cs);
                 qsa_block_topk(scores_ + (size_t) tb * max_blocks_, step_ + tb * kStepCount, n, max_blocks_, cap_, s,
@@ -649,6 +701,7 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
                 const QsaAttnPools pools = qsa_attn_pools(st);
                 qsa_decode_attn_batch(qcur_ + tb * NH * HD, pools, sel_ + (size_t) tb * cap_, step_ + tb * kStepCount, cap_,
                                       s, attn_scratch_ + (size_t) tb * attn_scratch_floats_, attn_ + tb * NH * HD, n, cs);
+                }
                 stamp(l, 13, grp);
                 if (st.kv_rot || st.kv_hybrid) fwht256_inplace_cuda(attn_ + tb * NH * HD, (int64_t) n * NH, cs);   // back: H^-1 = H
                 if (qb) native_qsa_gate_apply(attn_ + tb * NH * HD, qfull_ + tb * NH * 2 * HD, attn32_ + tb * NH * HD,
@@ -820,9 +873,10 @@ bool Verifier::record_window(int T, cudaStream_t cs, std::string& err) {
         }
     if (le_ < g.n_layers) {   // a layer split's earlier stage: hand the residual on, no head
         for (int t = 0; t < T; ++t) {
-            copy_from_mapped(hand_out_ + (size_t) t * HB, Rt(t), HC * N, cs);
-            copy_from_mapped(hand_out_ + (size_t) t * HB + HC * N, bo_ + (size_t) t * N, N, cs);
-            copy_from_mapped(hand_out_ + (size_t) t * HB + HC * N + N, inj2_ + (size_t) t * HC, HC, cs);
+            const size_t hr = (size_t) (hrow0 + t) * HB;
+            copy_from_mapped(hand_out_ + hr, Rt(t), HC * N, cs);
+            copy_from_mapped(hand_out_ + hr + HC * N, bo_ + (size_t) t * N, N, cs);
+            copy_from_mapped(hand_out_ + hr + HC * N + N, inj2_ + (size_t) t * HC, HC, cs);
         }
         return true;
     }
@@ -1019,10 +1073,45 @@ bool Verifier::capture_commit(std::string& err) {
     return true;
 }
 
+void Verifier::collect_profile() {
+    const ModelGeometry& g = *g_;
+    cudaMemcpy(prof_h_.data(), prof_, prof_h_.size() * 8, cudaMemcpyDeviceToHost);
+    const int64_t L = g.n_layers;
+    auto at = [&](int64_t l, int i) { return prof_h_[(size_t) (l * kProfPer + i)]; };
+    // D8: the derived columns below read stamps the hc-read kernels write themselves or the next
+    // layer's first.  A slot no kernel stamped is 0 and its unsigned difference wrapped to ~1e19 ns -
+    // which is why (gap)/head/hc0 printed ~1e14 ms per window.  A missing or out-of-order stamp now
+    // contributes nothing.
+    const auto gap = [](unsigned long long to, unsigned long long from) {
+        return (from != 0 && to != 0 && to >= from) ? (double) (to - from) : 0.0;
+    };
+    for (int64_t l = 0; l < L; ++l) {
+        const int kind = is_qsa_layer(g, l) ? 1 : 0;
+        unsigned long long prev = at(l, 0);
+        for (int i = 1; i <= 24; ++i) {
+            const unsigned long long x = at(l, i);
+            if (x == 0 || x < prev) continue;
+            prof_sum_[kind][i] += (double) (x - prev);
+            prev = x;
+        }
+        if (l + 1 < L) prof_sum_[kind][25] += gap(at(l + 1, 0), at(l, 24));
+        const double dn = gap(at(l, 27), at(l, 0)), dd = gap(at(l, 28), at(l, 27)), du = gap(at(l, 1), at(l, 28));
+        if (dn > 0 && dd > 0 && du > 0) {   // the split exists: show it split, not twice
+            prof_sum_[kind][27] += dn;      // hc-read0: norm
+            prof_sum_[kind][28] += dd;      //           down
+            prof_sum_[kind][29] += du;      //           up (through both halves, as before)
+            prof_sum_[kind][1] -= gap(at(l, 1), at(l, 0));   // (hc-read0 shown split)
+        }
+    }
+    prof_sum_[0][26] += gap(at(L, 1), at(L, 0));
+    ++prof_windows_;
+}
+
 bool Verifier::run(int T, const int32_t* tokens, int64_t pos0, PoolMultiFn pool, void* user, int32_t* out,
                    std::string& err) {
     using namespace strata::kernels;
     const OnDevice on_device(device_);
+    last_batch_ = false;
     if (T < 1 || T > max_t_) { err = "verify: window size out of range"; return false; }
     if (released_.load()) { err = "verify: an earlier window never finished on the GPU (#267); restart the engine"; return false; }
     const ModelGeometry& g = *g_;
@@ -1135,38 +1224,7 @@ bool Verifier::run(int T, const int32_t* tokens, int64_t pos0, PoolMultiFn pool,
     if (se != cudaSuccess) { err = std::string("verify: ") + cudaGetErrorString(se); return false; }
     progress_at("verify window: waiting for the expert copies", (int64_t) T);
     cudaStreamSynchronize(copy_);   // no host function of this window may raise flag B in the next one
-    if (prof_on_ && G == 1) {       // the window's GPU stage stamps
-        cudaMemcpy(prof_h_.data(), prof_, prof_h_.size() * 8, cudaMemcpyDeviceToHost);
-        const int64_t L = g.n_layers;
-        auto at = [&](int64_t l, int i) { return prof_h_[(size_t) (l * kProfPer + i)]; };
-        // D8: the derived columns below read stamps the hc-read kernels write themselves or the next
-        // layer's first.  A slot no kernel stamped is 0 and its unsigned difference wrapped to ~1e19 ns -
-        // which is why (gap)/head/hc0 printed ~1e14 ms per window.  A missing or out-of-order stamp now
-        // contributes nothing.
-        const auto gap = [](unsigned long long to, unsigned long long from) {
-            return (from != 0 && to != 0 && to >= from) ? (double) (to - from) : 0.0;
-        };
-        for (int64_t l = 0; l < L; ++l) {
-            const int kind = is_qsa_layer(g, l) ? 1 : 0;
-            unsigned long long prev = at(l, 0);
-            for (int i = 1; i <= 24; ++i) {
-                const unsigned long long x = at(l, i);
-                if (x == 0 || x < prev) continue;
-                prof_sum_[kind][i] += (double) (x - prev);
-                prev = x;
-            }
-            if (l + 1 < L) prof_sum_[kind][25] += gap(at(l + 1, 0), at(l, 24));
-            const double dn = gap(at(l, 27), at(l, 0)), dd = gap(at(l, 28), at(l, 27)), du = gap(at(l, 1), at(l, 28));
-            if (dn > 0 && dd > 0 && du > 0) {   // the split exists: show it split, not twice
-                prof_sum_[kind][27] += dn;      // hc-read0: norm
-                prof_sum_[kind][28] += dd;      //           down
-                prof_sum_[kind][29] += du;      //           up (through both halves, as before)
-                prof_sum_[kind][1] -= gap(at(l, 1), at(l, 0));   // (hc-read0 shown split)
-            }
-        }
-        prof_sum_[0][26] += gap(at(L, 1), at(L, 0));
-        ++prof_windows_;
-    }
+    if (prof_on_ && G == 1) collect_profile();   // the window's GPU stage stamps
     // ---- a sampled or penalized request: the head's sampling again, host-side so its parameters are this call's
     // own (a captured kernel would replay the same draws forever).  Row t's draw is Philox(seed, pos0 + t): tied to
     // the POSITION it samples, not to how the text was cut into windows, so a seed replays the same text whatever
@@ -1221,7 +1279,7 @@ void Verifier::set_plan_slot(int grp) {
     sink_.ptr = (unsigned long long*) (base + ptr_off);
     sink_.ptr2 = sink_.ptr + cap;
     sink_.start2 = base + ptr_off + 4 * cap;
-    const int G = groups_[last_t_] > 0 ? groups_[last_t_] : 1;
+    const int G = last_batch_ ? 1 : (groups_[last_t_] > 0 ? groups_[last_t_] : 1);
     const int64_t per = G == 2 ? kStagingBlobs / 2 : kStagingBlobs;
     sink_.staging = (unsigned long long) (staging_ + (size_t) (grp * per) * strata::kernels::cpu::expert_layout().max_blob);
     sink_.staging_cap = per;
@@ -1363,6 +1421,394 @@ bool Verifier::wait_commit(std::string& err) {
     return next_ == nullptr || next_->wait_commit(err);
 }
 
+
+// ================================ BATCH WINDOWS (see init_slots) ================================
+
+bool Verifier::init_slots(const std::vector<SessionState*>& slots, std::string& err) {
+    const OnDevice on_device(device_);
+    if (g_ == nullptr || ss_ == nullptr) { err = "verify: init_slots before init"; return false; }
+    if (slots.empty() || (int) slots.size() > max_t_) {
+        err = "verify: init_slots needs 1.." + std::to_string(max_t_) + " sessions";
+        return false;
+    }
+    for (SessionState* x : slots) {
+        if (x == nullptr || x->layer_lo != ss_->layer_lo || x->layer_hi != ss_->layer_hi ||
+            x->max_cells != ss_->max_cells || x->gdn_ord0 != ss_->gdn_ord0 || x->qsa_ord0 != ss_->qsa_ord0) {
+            err = "verify: a slot's session is not carved like the stage's own (layer range, context)";
+            return false;
+        }
+    }
+    const strata::kernels::QsaShapes s = shapes_of(*g_);
+    const int64_t S = (int64_t) slots.size(), CB = 2 + max_t_;
+    const int64_t TS = (s.idx_block - 1) * g_->idx_key_dim, nQ = g_->n_qsa_layers();
+    void* d = nullptr;
+    if (!mapped((size_t) (S * CB * 4 + 16), (void**) &h_commitb_, (void**) &m_commitb_)) {
+        err = "verify: the batch commit staging failed";
+        return false;
+    }
+    const uint64_t a = ((uint64_t) S * CB * 4 + 255) & ~255ull;
+    if (cudaMalloc(&d, a + (uint64_t) S * std::max<int64_t>(nQ, 1) * TS * 4) != cudaSuccess) {
+        err = "verify: the batch buffers do not fit";
+        return false;
+    }
+    arena_b_ = d;
+    commitb_ = (int32_t*) d;
+    tail_snap_b_ = (float*) ((uint8_t*) d + a);
+    slots_ = slots;
+    slot_sp_.assign(slots.size(), sampling_);   // greedy until set_slot_sampling
+    std::fprintf(stderr, "strata verify: batch windows of up to %lld sequences (layers [%lld, %lld))\n", (long long) S,
+                 (long long) lb_, (long long) le_);
+    return true;
+}
+
+bool Verifier::capture_batch(int base, int S, std::string& err) {
+    cudaGraphExec_t& ex = exec_bm_[base * 16 + S];
+    if (ex != nullptr) return true;
+    if (cudaStreamBeginCapture(cs_, cudaStreamCaptureModeThreadLocal) != cudaSuccess) {
+        err = "verify: begin batch capture failed";
+        return false;
+    }
+    batch_rec_ = true;
+    row_base_ = base;
+    std::string rerr;
+    const bool ok = record_window(S, cs_, rerr);
+    batch_rec_ = false;
+    row_base_ = 0;
+    cudaGraph_t graph = nullptr;
+    const cudaError_t ce = cudaStreamEndCapture(cs_, &graph);
+    if (!ok || ce != cudaSuccess) {
+        if (graph) cudaGraphDestroy(graph);
+        err = !ok ? rerr : std::string("verify: end batch capture: ") + cudaGetErrorString(ce);
+        return false;
+    }
+    const cudaError_t ie = cudaGraphInstantiate(&ex, graph, 0);
+    cudaGraphDestroy(graph);
+    if (ie != cudaSuccess) { err = std::string("verify: batch instantiate: ") + cudaGetErrorString(ie); return false; }
+    cudaGraphUpload(ex, cs_);
+    cudaStreamSynchronize(cs_);
+    std::fprintf(stderr, "strata verify: captured the batch window over slots [%d, %d)\n", base, base + S);
+    return true;
+}
+
+bool Verifier::capture_commit_batch(int base, int S, std::string& err) {
+    cudaGraphExec_t& cex = commit_bm_[base * 16 + S];
+    if (cex != nullptr) return true;
+    using namespace strata::kernels;
+    const ModelGeometry& g = *g_;
+    const QsaShapes s = shapes_of(g);
+    const int64_t C = g.ssm_conv_channels, HV = g.ssm_v_heads, ZV = g.ssm_value_dim, ID = g.idx_key_dim, MT = max_t_;
+    const int64_t CB = 2 + MT, nQ = g.n_qsa_layers();
+    const uint64_t gdn_floats = (uint64_t) g.ssm_state_size * g.ssm_v_heads * g.ssm_state_size +
+                                (uint64_t) g.ssm_conv_channels * (g.ssm_d_conv - 1);
+    const int64_t TS = (s.idx_block - 1) * ID;
+    const int64_t HS = (int64_t) NG_HIST * NG_HC_DIM;
+    if (cudaStreamBeginCapture(cs_, cudaStreamCaptureModeThreadLocal) != cudaSuccess) {
+        err = "verify: begin batch commit capture failed";
+        return false;
+    }
+    bool ok = true;
+    try {
+        copy_i32_from_mapped(commitb_ + (size_t) base * CB, m_commitb_ + (size_t) base * CB, S * CB, cs_);
+        int64_t qsa_index = 0, gdn_index = 0;
+        for (int64_t l = 0; l < lb_; ++l) (is_qsa_layer(g, l) ? qsa_index : gdn_index) += 1;
+        for (int64_t l = lb_; l < le_ && ok; ++l) {
+            const LayerView v(*wt_, l);
+            if (!is_qsa_layer(g, l)) {
+                const WeightRef* wnm = need(v, "ssm_norm.weight", err);
+                if (!wnm) { ok = false; break; }
+                for (int t = 0; t < S; ++t) {
+                    SessionState& sx = *slots_[(size_t) (base + t)];
+                    const int32_t* keep = commitb_ + (size_t) (base + t) * CB;
+                    float* state = sx.gdn_state + (size_t) (gdn_index - sx.gdn_ord0) * gdn_floats;
+                    float* conv = state + (uint64_t) g.ssm_state_size * g.ssm_v_heads * g.ssm_state_size;
+                    gdn_conv_commit(conv, qkv_L_ + (size_t) gdn_index * MT * C + (size_t) t * C, (int) C, keep, cs_);
+                    gdn_step_norm_multi(state, h_L_ + (size_t) gdn_index * MT * C + (size_t) t * C, (int) C,
+                                        gate_L_ + (size_t) gdn_index * MT * HV + (size_t) t * HV,
+                                        beta_L_ + (size_t) gdn_index * MT * HV + (size_t) t * HV, z_ + (size_t) t * ZV,
+                                        (const float*) wnm->data, EPS, y_dummy_ + (size_t) t * ZV, (int) g.ssm_k_heads,
+                                        (int) HV, 1, keep, cs_);
+                }
+                ++gdn_index;
+            } else {
+                const WeightRef* wikn = need(v, "indexer.k_norm.weight", err);
+                if (!wikn) { ok = false; break; }
+                for (int t = 0; t < S; ++t) {
+                    const QsaState& st = slots_[(size_t) (base + t)]->qsa_states[qsa_index];
+                    copy_from_mapped(st.idx_tail, tail_snap_b_ + ((size_t) (base + t) * nQ + qsa_index) * TS, TS, cs_);
+                    const QsaIndexerBuffers ib{st.idx_tail, st.idx_dead, st.idx_pooled, st.idx_block_pos};
+                    native_qsa_indexer_append(idx_raw_L_ + (size_t) (qsa_index * MT + t) * ID, commitb_ + (size_t) (base + t) * CB + 2,
+                                              0, (const float*) wikn->data, EPS, ib, s, st.max_cells, rope_scaling(), cs_);
+                }
+                ++qsa_index;
+            }
+        }
+        if (ok && ss_->ple.ready() && ple_stage())
+            for (int t = 0; t < S; ++t)
+                copy_indexed(slots_[(size_t) (base + t)]->ple_hist, hist_snap_ + (size_t) t * HS, HS,
+                             commitb_ + (size_t) (base + t) * CB + 1, HS, cs_);
+    } catch (const std::exception& e) {
+        err = std::string("verify batch commit: ") + e.what();
+        ok = false;
+    }
+    cudaGraph_t graph = nullptr;
+    const cudaError_t ce = cudaStreamEndCapture(cs_, &graph);
+    if (!ok) {
+        if (graph) cudaGraphDestroy(graph);
+        return false;
+    }
+    if (ce != cudaSuccess || cudaGraphInstantiate(&cex, graph, 0) != cudaSuccess) {
+        if (graph) cudaGraphDestroy(graph);
+        err = std::string("verify: batch commit capture: ") + cudaGetErrorString(ce);
+        return false;
+    }
+    cudaGraphDestroy(graph);
+    return true;
+}
+
+bool Verifier::stage_batch(int base, int S, const int32_t* tokens, const int64_t* pos, std::string& err) {
+    using namespace strata::kernels;
+    if (S < 1 || base < 0 || base + S > (int) slots_.size()) { err = "verify: batch rows out of range (init_slots)"; return false; }
+    if (released_.load()) { err = "verify: an earlier window never finished on the GPU (#267); restart the engine"; return false; }
+    const ModelGeometry& g = *g_;
+    for (int t = 0; t < S; ++t)
+        if (pos[t] < 0 || pos[t] + 1 > slots_[(size_t) (base + t)]->max_cells) {
+            err = "verify: slot " + std::to_string(base + t) + " runs past its context";
+            return false;
+        }
+    if (!capture_batch(base, S, err) || !capture_commit_batch(base, S, err)) return false;
+    const Clock::time_point t0 = Clock::now();
+    const QsaShapes s = shapes_of(g);
+    for (int t = 0; t < S; ++t) {
+        h_tok_[t] = tokens[t];
+        qsa_step_fill(h_step_ + t * kStepCount, pos[t], s);
+        for (int64_t h = 0; h < g.n_head; ++h) h_pos_[t * g.n_head + h] = (int32_t) pos[t];
+        int32_t* pk = h_pos_ + (size_t) max_t_ * g.n_head;
+        int32_t* pi = pk + (size_t) max_t_ * g.n_head_kv;
+        for (int64_t h = 0; h < g.n_head_kv; ++h) pk[t * g.n_head_kv + h] = (int32_t) pos[t];
+        for (int64_t h = 0; h < g.idx_q_heads; ++h) pi[t * g.idx_q_heads + h] = (int32_t) pos[t];
+    }
+    if (ss_->ple.ready() && ple_stage()) {
+        uint32_t rows[kVerifyMaxT * PLE_N_HEADS];
+        for (int t = 0; t < S; ++t) {
+            const SessionState& sx = *slots_[(size_t) (base + t)];
+            int32_t prev[2] = {sx.ple_prev[0], sx.ple_prev[1]};
+            ngram_rows(&tokens[t], prev, 1, ss_->ple.consts, rows + t * PLE_N_HEADS);
+        }
+        if (!ss_->ple.table->gather_batch(rows, (size_t) S, h_ple_, err)) return false;
+    }
+    // the commit's rows (a batch window keeps every row): [1, 0, position, -1 ..] per slot
+    const int64_t CB = 2 + max_t_;
+    for (int t = 0; t < S; ++t) {
+        int32_t* c = h_commitb_ + (size_t) (base + t) * CB;
+        c[0] = 1;
+        c[1] = 0;
+        c[2] = (int32_t) pos[t];
+        for (int64_t j = 3; j < CB; ++j) c[j] = -1;
+    }
+    *(volatile uint32_t*) h_seq_ = 0;
+    *(volatile uint32_t*) h_flag_ = 0;
+    *(volatile uint32_t*) h_flagA_ = 0;
+    *(volatile uint32_t*) h_flagB_ = 0;
+    std::atomic_thread_fence(std::memory_order_seq_cst);
+    last_t_ = S;
+    last_base_ = base;
+    last_batch_ = true;
+    for (int t = 0; t < S; ++t) { last_tokens_[t] = tokens[t]; last_pos_b_[t] = pos[t]; }
+    ms_host += ms_since(t0);
+    return true;
+}
+
+bool Verifier::run_slots(int S, const int32_t* tokens, const int64_t* pos, PoolMultiFn pool, void* user, int32_t* out,
+                         std::string& err) {
+    using namespace strata::kernels;
+    const OnDevice on_device(device_);
+    const ModelGeometry& g = *g_;
+    if (!stage_batch(0, S, tokens, pos, err)) return false;
+    const cudaError_t le = cudaGraphLaunch(exec_bm_[0 * 16 + S], cs_);
+    if (le != cudaSuccess) { err = std::string("verify: batch launch: ") + cudaGetErrorString(le); return false; }
+    (void) cudaStreamQuery(cs_);
+    volatile uint32_t* const seq = h_seq_;
+    volatile uint32_t* const flag = h_flag_;
+    const int64_t steps = le_ - lb_;
+    for (int64_t k = 0; k < steps; ++k) {
+        const int64_t l = lb_ + k;
+        const uint32_t want = (uint32_t) (k + 1);
+        const Clock::time_point a = Clock::now();
+        auto last_flush = a;
+        uint32_t spins = 0;
+        progress_at("verify batch: waiting for the GPU to reach layer", l);
+        while (*seq < want) {
+            _mm_pause();
+            if ((++spins & 1023u) != 0) continue;
+            const auto now = Clock::now();
+            if (now - last_flush > std::chrono::microseconds(2000)) {
+                last_flush = now;
+                const cudaError_t q = cudaStreamQuery(cs_);
+                if (q != cudaErrorNotReady && *seq < want) {
+                    err = "verify batch: layer " + std::to_string(l) + " never rang (" +
+                          (q == cudaSuccess ? std::string("graph finished") : std::string(cudaGetErrorString(q))) + ")";
+                    return false;
+                }
+            }
+            if (now - a > std::chrono::seconds(20)) {
+                err = "verify batch: timed out at layer " + std::to_string(l) + released_note(release_gpu_waits(5000));
+                return false;
+            }
+        }
+        const Clock::time_point b = Clock::now();
+        cur_layer_ = want - 1;
+        set_plan_slot(0);
+        progress_at("verify batch: the CPU experts of layer", l);
+        if (pool != nullptr) pool(user, h_x_, h_ids_, S, ss_->k, h_ymiss_, l);
+        progress_tick();
+        std::atomic_thread_fence(std::memory_order_seq_cst);
+        _mm_sfence();
+        if (*(volatile uint32_t*) h_flagA_ != want) {        // the pool did not publish a plan: an empty one
+            sink_.counts[0] = 0;
+            sink_.counts[1] = 0;
+            sink_.counts[2] = 0;
+            sink_.start[0] = 0;
+            sink_.start2[0] = 0;
+            std::atomic_thread_fence(std::memory_order_seq_cst);
+            *(volatile uint32_t*) h_flagA_ = want;
+            raise_flag(h_flagB_, want);
+        }
+        *flag = want;
+        ms_wait += std::chrono::duration<double, std::milli>(b - a).count();
+        ms_pool += ms_since(b);
+    }
+    const cudaError_t se = cudaStreamSynchronize(cs_);
+    if (se != cudaSuccess) { err = std::string("verify batch: ") + cudaGetErrorString(se); return false; }
+    cudaStreamSynchronize(copy_);
+    if (prof_on_) collect_profile();
+    ++windows;
+    if (le_ < g.n_layers) return next_ == nullptr || next_->run_slots(S, tokens, pos, pool, next_user_, out, err);
+    if (!sample_rows(0, S, err)) return false;
+    for (int t = 0; t < S; ++t) out[t] = ((volatile int32_t*) h_out_)[t];
+    progress_at("decode");
+    progress_beat();
+    return true;
+}
+
+bool Verifier::commit_slots(std::string& err) {
+    const OnDevice on_device(device_);
+    if (!last_batch_ || last_t_ < 1) { err = "verify: commit_slots without a batch window"; return false; }
+    const int S = last_t_;
+    const Clock::time_point t0 = Clock::now();
+    std::atomic_thread_fence(std::memory_order_seq_cst);
+    const cudaError_t le = cudaGraphLaunch(commit_bm_[last_base_ * 16 + S], cs_);
+    if (le != cudaSuccess) { err = std::string("verify: batch commit launch: ") + cudaGetErrorString(le); return false; }
+    const cudaError_t se = cudaStreamSynchronize(cs_);
+    if (se != cudaSuccess) { err = std::string("verify: batch commit: ") + cudaGetErrorString(se); return false; }
+    if (ple_stage())
+        for (int t = 0; t < S; ++t) {
+            SessionState& sx = *slots_[(size_t) (last_base_ + t)];
+            sx.ple_prev[0] = sx.ple_prev[1];
+            sx.ple_prev[1] = last_tokens_[t];
+        }
+    ms_commit += ms_since(t0);
+    return next_ == nullptr || next_->commit_slots(err);
+}
+
+bool Verifier::sample_rows(int base, int S, std::string& err) {
+    bool any = false;
+    for (int t = 0; t < S; ++t) {
+        strata::kernels::SamplerParams sp = slot_sp_[(size_t) (base + t)];
+        if (sp.greedy || sp.temperature <= 0.0f) continue;
+        sp.counter = (uint64_t) last_pos_b_[t];   // Philox(seed, position): the solo window's draw for this position
+        sp.penalty_last_n = 0;
+        strata::kernels::sample_tokens(head_logits_ + (size_t) t * (size_t) n_vocab_, 1, (int) n_vocab_, nullptr, 0, sp,
+                                       m_out_ + t, cs_);
+        any = true;
+    }
+    if (any && cudaStreamSynchronize(cs_) != cudaSuccess) { err = "verify batch: the row sampling failed"; return false; }
+    return true;
+}
+
+bool Verifier::batch_launch(int base, int S, const int32_t* tokens, const int64_t* pos, std::string& err) {
+    const OnDevice on_device(device_);
+    if (b_running_) { err = "verify: batch_launch while this stage is busy"; return false; }
+    if (!stage_batch(base, S, tokens, pos, err)) return false;
+    cudaError_t le = cudaGraphLaunch(exec_bm_[base * 16 + S], cs_);
+    if (le == cudaSuccess) le = cudaGraphLaunch(commit_bm_[base * 16 + S], cs_);   // right behind it: every row is kept
+    if (le != cudaSuccess) { err = std::string("verify: batch launch: ") + cudaGetErrorString(le); return false; }
+    (void) cudaStreamQuery(cs_);
+    if (ple_stage())   // the host's side of the commit (the hash's last two tokens)
+        for (int t = 0; t < S; ++t) {
+            SessionState& sx = *slots_[(size_t) (base + t)];
+            sx.ple_prev[0] = sx.ple_prev[1];
+            sx.ple_prev[1] = tokens[t];
+        }
+    b_running_ = true;
+    b_k_ = 0;
+    b_steps_ = le_ - lb_;
+    b_last_ = Clock::now();
+    return true;
+}
+
+int Verifier::batch_poll(PoolMultiFn pool, void* user, std::string& err) {
+    if (!b_running_) return 1;
+    const OnDevice on_device(device_);
+    volatile uint32_t* const seq = h_seq_;
+    const int S = last_t_;
+    while (b_k_ < b_steps_) {
+        const uint32_t want = (uint32_t) (b_k_ + 1);
+        if (*seq < want) {
+            const auto now = Clock::now();
+            if (now - b_last_ > std::chrono::milliseconds(2)) {
+                const cudaError_t q = cudaStreamQuery(cs_);
+                if (q != cudaErrorNotReady && *seq < want) {
+                    err = "verify batch: layer " + std::to_string(lb_ + b_k_) + " never rang (" +
+                          (q == cudaSuccess ? std::string("graph finished") : std::string(cudaGetErrorString(q))) + ")";
+                    b_running_ = false;
+                    return -1;
+                }
+                if (now - b_last_ > std::chrono::seconds(20)) {
+                    err = "verify batch: timed out at layer " + std::to_string(lb_ + b_k_) + released_note(release_gpu_waits(5000));
+                    b_running_ = false;
+                    return -1;
+                }
+            }
+            return 0;
+        }
+        const Clock::time_point b = Clock::now();
+        cur_layer_ = want - 1;
+        set_plan_slot(0);
+        if (pool != nullptr) pool(user, h_x_, h_ids_, S, ss_->k, h_ymiss_, lb_ + b_k_);
+        progress_tick();
+        std::atomic_thread_fence(std::memory_order_seq_cst);
+        _mm_sfence();
+        if (*(volatile uint32_t*) h_flagA_ != want) {        // the pool did not publish a plan: an empty one
+            sink_.counts[0] = 0;
+            sink_.counts[1] = 0;
+            sink_.counts[2] = 0;
+            sink_.start[0] = 0;
+            sink_.start2[0] = 0;
+            std::atomic_thread_fence(std::memory_order_seq_cst);
+            *(volatile uint32_t*) h_flagA_ = want;
+            raise_flag(h_flagB_, want);
+        }
+        *(volatile uint32_t*) h_flag_ = want;
+        ms_pool += ms_since(b);
+        b_last_ = Clock::now();
+        ++b_k_;
+    }
+    const cudaError_t q = cudaStreamQuery(cs_);
+    if (q == cudaErrorNotReady) return 0;
+    if (q != cudaSuccess) { err = std::string("verify batch: ") + cudaGetErrorString(q); b_running_ = false; return -1; }
+    const cudaError_t qc = cudaStreamQuery(copy_);   // no host function of this window may raise flag B in the next
+    if (qc == cudaErrorNotReady) return 0;
+    if (prof_on_) collect_profile();
+    if (last_stage()) {
+        if (!sample_rows(last_base_, S, err)) { b_running_ = false; return -1; }
+        for (int t = 0; t < S; ++t) b_out_[t] = ((volatile int32_t*) h_out_)[t];
+    }
+    ++windows;
+    b_running_ = false;
+    progress_beat();
+    return 1;
+}
+
 bool Verifier::copy_logits(int t, float* host) const {
     if (next_ != nullptr) return next_->copy_logits(t, host);   // a layer split: the head is on the last stage
     if (head_logits_ == nullptr || host == nullptr || t < 0 || n_vocab_ <= 0) return false;

src/program/generate.cpp

@@ -363,6 +363,16 @@ struct Options {
     /// spec-1 drafts); 0 = plain decode.  `spec_oracle` drafts from a token file (the expected continuation, for
     /// the exactness test); `spec_corrupt` N > 0 replaces every Nth draft with a wrong token.
     int spec = 0;
+    /// --serve runs up to this many sequences at once, one token each per batch window (0 = off).  A request
+    /// sent as `BGEN <slot> <max_new> ...` reads its prompt and its first token through the usual path, then
+    /// continues in slot <slot> of the batch windows (`BT <slot> <id>` lines, then `BDONE <slot> ...`).
+    int batch = 0;
+    /// With an explicit --layer-split, every GPU loads only its own layers' dense weights instead of the whole
+    /// model's: the VRAM they took goes back to the expert cache (opt-in)
+    bool trim_stage_weights = false;
+    /// The --batch slots in this many groups pipelined through the stages of a layer split (stage k
+    /// runs one group while stage k+1 runs another).  1 = every slot in one window, stage after stage.
+    int batch_groups = 1;
     std::string spec_oracle;
     int spec_corrupt = 0;
     /// Plan v0.3 P6: the MTP draft layer's runtime directory (tools/mtp_rt.py); drafts come from it.
@@ -1214,6 +1224,9 @@ int main(int argc, char** argv) {
         else if (a == "--prefill-until") o.prefill_until = std::atoll(next("--prefill-until"));
         else if (a == "--dump-final-r") o.dump_final_r = next("--dump-final-r");
         else if (a == "--spec") o.spec = std::atoi(next("--spec"));
+        else if (a == "--batch") o.batch = std::atoi(next("--batch"));
+        else if (a == "--trim-stage-weights") o.trim_stage_weights = true;
+        else if (a == "--batch-groups") o.batch_groups = std::atoi(next("--batch-groups"));
         else if (a == "--spec-oracle") o.spec_oracle = next("--spec-oracle");
         else if (a == "--spec-corrupt") o.spec_corrupt = std::atoi(next("--spec-corrupt"));
         else if (a == "--mtp") o.mtp = next("--mtp");
@@ -1984,8 +1997,28 @@ int main(int argc, char** argv) {
         }
         if (native_pack) skip.insert("token_embd.weight");
     }
+    // --trim-stage-weights: the layers each device runs, known before anything loads (explicit split only)
+    const bool trim = o.trim_stage_weights && multi_gpu && !split_auto;
+    if (o.trim_stage_weights && !trim)
+        std::fprintf(stderr, "strata generate: --trim-stage-weights needs an explicit --layer-split (ignored)\n");
+    std::vector<std::string> pack_names;
+    if (trim) {
+        std::ifstream idx(o.pack + "/index.txt");
+        for (std::string ln; std::getline(idx, ln);)
+            if (!ln.empty() && ln[0] != '#') pack_names.push_back(ln.substr(0, ln.find(' ')));
+    }
+    auto range_skip = [&](int64_t lo, int64_t hi) {   // skip + every blk.<l>. tensor outside [lo, hi)
+        std::set<std::string> out = skip;
+        for (const std::string& nm : pack_names) {
+            if (nm.rfind("blk.", 0) != 0) continue;
+            const long l = std::strtol(nm.c_str() + 4, nullptr, 10);
+            if (l < lo || l >= hi) out.insert(nm);
+        }
+        return out;
+    };
+    const std::set<std::string> skip0 = trim ? range_skip(0, split_at[0]) : skip;
     uint64_t pool_bytes = 0;
-    if (!strata::core::WeightTable::pool_bytes(o.pack, pool_bytes, err, skip.empty() ? nullptr : &skip)) {
+    if (!strata::core::WeightTable::pool_bytes(o.pack, pool_bytes, err, skip0.empty() ? nullptr : &skip0)) {
         std::fprintf(stderr, "strata generate: %s\n", err.c_str());
         return 1;
     }
@@ -2005,7 +2038,7 @@ int main(int argc, char** argv) {
         return 1;
     }
     strata::core::WeightTable wt;
-    if (!wt.load(o.pack, arena, pool_bytes, err, skip.empty() ? nullptr : &skip)) {
+    if (!wt.load(o.pack, arena, pool_bytes, err, skip0.empty() ? nullptr : &skip0)) {
         std::fprintf(stderr, "strata generate: %s\n", err.c_str());
         return 1;
     }
@@ -2015,7 +2048,7 @@ int main(int argc, char** argv) {
 
     strata::core::NativeDense native_dense;
     if (!o.native_dense_gguf.empty()) {
-        if (!native_dense.load(o.native_dense_gguf, wt, err, o.native_ple_key)) {
+        if (!native_dense.load(o.native_dense_gguf, wt, err, o.native_ple_key, 0, trim ? split_at[0] : -1)) {
             std::fprintf(stderr, "strata generate: native dense projections: %s\n", err.c_str());
             return 1;
         }
@@ -2304,19 +2337,27 @@ int main(int argc, char** argv) {
         }
         const strata::core::OnDevice on(st.dev);
         void* arena_s = nullptr;
-        if (cudaMalloc(&arena_s, pool_bytes) != cudaSuccess ||
-            !st.wt.load(o.pack, arena_s, pool_bytes, err, skip.empty() ? nullptr : &skip)) {
+        // --trim-stage-weights: this stage's layers only
+        const int64_t s_lo = trim ? split_at[i] : 0, s_hi = trim ? (i + 1 < split_at.size() ? split_at[i + 1] : g.n_layers) : -1;
+        const std::set<std::string> skip_s = trim ? range_skip(s_lo, s_hi) : skip;
+        uint64_t pool_s = pool_bytes;
+        if (trim && !strata::core::WeightTable::pool_bytes(o.pack, pool_s, err, &skip_s)) {
+            std::fprintf(stderr, "strata generate: layer split, CUDA%d: %s\n", st.dev, err.c_str());
+            return 1;
+        }
+        if (cudaMalloc(&arena_s, pool_s) != cudaSuccess ||
+            !st.wt.load(o.pack, arena_s, pool_s, err, skip_s.empty() ? nullptr : &skip_s)) {
             cudaGetLastError();
             size_t free_b = 0, total_b = 0;   // #486: what that card had free
             cudaMemGetInfo(&free_b, &total_b);
             std::fprintf(stderr, "strata generate: layer split, CUDA%d weights: %s (%llu MiB needed, %llu MiB of %llu "
                                  "MiB free on that card)\n", st.dev,
                          err.empty() ? "the weight arena does not fit" : err.c_str(),
-                         (unsigned long long) (pool_bytes >> 20), (unsigned long long) (free_b >> 20),
+                         (unsigned long long) (pool_s >> 20), (unsigned long long) (free_b >> 20),
                          (unsigned long long) (total_b >> 20));
             return 1;
         }
-        if (!o.native_dense_gguf.empty() && !st.dense.load(o.native_dense_gguf, st.wt, err, o.native_ple_key)) {
+        if (!o.native_dense_gguf.empty() && !st.dense.load(o.native_dense_gguf, st.wt, err, o.native_ple_key, s_lo, s_hi)) {
             std::fprintf(stderr, "strata generate: layer split, CUDA%d native dense projections: %s\n", st.dev,
                          err.c_str());
             return 1;
@@ -2648,6 +2689,40 @@ int main(int argc, char** argv) {
                      last ? " and the head" : "", (double) fb / 1073741824.0);
     }
 
+    // --batch: every stage carves o.batch more sessions like its own (same layer range and context), one per
+    // slot of the batch windows.  Before the expert cache is sized, so `--expert-cache auto` leaves them room.
+    std::vector<std::vector<std::unique_ptr<strata::core::SessionState>>> bslot_ss;
+    if (o.batch > 0) {
+        if (!o.serve || o.batch < 2 || o.batch > strata::kernels::kVerifyMaxT) {
+            std::fprintf(stderr, "strata generate: --batch N needs --serve and 2 <= N <= %d\n", strata::kernels::kVerifyMaxT);
+            return 2;
+        }
+        bslot_ss.resize(1 + stages.size());
+        for (size_t k = 0; k < bslot_ss.size(); ++k) {
+            const int dev = k == 0 ? 0 : stages[k - 1]->dev;
+            const int64_t lo = k == 0 ? 0 : stages[k - 1]->lb;
+            const int64_t hi = k == 0 ? (multi_gpu ? split_at[0] : -1) : stages[k - 1]->le;
+            const strata::core::OnDevice on_k(dev);
+            const uint64_t bytes = strata::core::session_bytes(g, o.max_context, K, lo, hi);
+            for (int b = 0; b < o.batch; ++b) {
+                auto u = std::make_unique<strata::core::SessionState>();
+                void* buf = nullptr;
+                if (cudaMalloc(&buf, bytes) != cudaSuccess || strata::core::session_init(g, o.max_context, K, buf, *u, lo, hi) == 0) {
+                    std::fprintf(stderr, "strata generate: --batch: slot %d's session on CUDA%d does not fit (%.2f GiB each)\n",
+                                 b, dev, (double) bytes / 1073741824.0);
+                    return 1;
+                }
+                strata::core::session_zero(*u, g, nullptr, nullptr);
+                bslot_ss[k].push_back(std::move(u));
+            }
+            cudaDeviceSynchronize();
+            size_t fb = 0, tb = 0;
+            cudaMemGetInfo(&fb, &tb);
+            std::fprintf(stderr, "strata generate: --batch: %d slot sessions on CUDA%d (%.2f GiB each); %.2f GiB free\n",
+                         o.batch, dev, (double) bytes / 1073741824.0, (double) fb / 1073741824.0);
+        }
+    }
+
     // Secure MTP's CUDA0 allocations before the large host arena is registered with both CUDA contexts.
     // In particular WDDM can refuse the draft weights after mapping tens of GiB of host pages.
     strata::core::MtpDrafter mtp;
@@ -4495,7 +4570,7 @@ int main(int argc, char** argv) {
             for (int st = 1; st < n_stages; ++st) {
                 bool ok_s = false;
                 if (split_same) {
-                    ok_s = ver_same.init(wt, g, ss, vh, native_head.loaded() ? &native_head : nullptr, o.spec, err);
+                    ok_s = ver_same.init(wt, g, ss, vh, native_head.loaded() ? &native_head : nullptr, std::max(o.spec, o.batch), err);
                 } else {
                     GpuStage& gs = *stages[(size_t) st - 1];
                     const strata::core::OnDevice on(gs.dev);
@@ -4505,7 +4580,7 @@ int main(int argc, char** argv) {
                     vs.blob = thits.blob;
                     vs.slot_off = gs.cache.slot_offsets();
                     vs.n_slots = gs.cache.slots();
-                    ok_s = gs.ver.init(gs.wt, g, gs.ss, vs, gs.head.loaded() ? &gs.head : nullptr, o.spec, err);
+                    ok_s = gs.ver.init(gs.wt, g, gs.ss, vs, gs.head.loaded() ? &gs.head : nullptr, std::max(o.spec, o.batch), err);
                     split_drive.cache_base[st] = gs.cache.device_slot(0);
                     split_drive.cache_slot_off[st] = gs.cache.slot_offsets();
                     split_drive.pcie_num[st] = pcie_num_of(gs.pcie_frac);
@@ -4522,11 +4597,28 @@ int main(int argc, char** argv) {
                           " (CUDA" + std::to_string(split_same ? 0 : stages[(size_t) st - 1]->dev) + ")";
             std::fprintf(stderr, "strata serve: layer split: layers %s, one hand-off per window\n", plan_s.c_str());
         }
-        if (!ver.init(wt, g, ss, vh, native_head.loaded() ? &native_head : nullptr, o.spec, err) ||
+        if (!ver.init(wt, g, ss, vh, native_head.loaded() ? &native_head : nullptr, std::max(o.spec, o.batch), err) ||
             !mtp.bind(last_st ? last_st->wt : wt, last_st ? &last_st->head : &native_head, ver.final_R_all(), err)) {
             std::fprintf(stderr, "strata serve: %s\n", err.c_str());
             return 1;
         }
+        // each stage's verifier gets its stage's slot sessions
+        if (o.batch > 0) {
+            if (split_same) {
+                std::fprintf(stderr, "strata serve: --batch needs each stage on its own GPU\n");
+                return 1;
+            }
+            for (size_t k = 0; k < bslot_ss.size(); ++k) {
+                std::vector<strata::core::SessionState*> ptrs;
+                for (auto& u : bslot_ss[k]) ptrs.push_back(u.get());
+                strata::core::Verifier& vk = k == 0 ? ver : stages[k - 1]->ver;
+                const strata::core::OnDevice on_k(k == 0 ? 0 : stages[k - 1]->dev);
+                if (!vk.init_slots(ptrs, err)) {
+                    std::fprintf(stderr, "strata serve: --batch, stage %zu: %s\n", k + 1, err.c_str());
+                    return 1;
+                }
+            }
+        }
         for (int st = 0; st < n_stages && n_stages > 1; ++st) {
             split_drive.plan[st] = stage_ver(st).plan_sink();
             if (st > 0) {
@@ -5044,7 +5136,290 @@ int main(int argc, char** argv) {
         bool mrope_identity = true;
         std::vector<float> img_rows;
         std::vector<const float*> row_ptr;
-        while (next_line(line)) {
+        // ---- --batch: the slots of the batch windows
+        struct BSlot {
+            bool active = false, stop = false;
+            int32_t x = 0;                 ///< the token the next window feeds (not yet in the slot's state)
+            int64_t p = 0;                 ///< its position
+            int64_t produced = 0, max_new = 0;
+            Clock::time_point t0;
+        };
+        std::vector<BSlot> bs((size_t) std::max(o.batch, 0));
+        // timing of the batch windows since the slots were last all idle (one stderr line then)
+        double bt_run = 0, bt_commit = 0, bt_emit = 0;
+        double bt_wait0 = 0, bt_pool0 = 0;
+        int64_t bt_miss0 = 0, bt_hits0 = 0, bt_pcie0 = 0;
+        int64_t bt_windows = 0, bt_rows = 0, bt_tokens = 0;
+        Clock::time_point bt_start = Clock::now();
+        int admit_slot = -1;               ///< the slot the request being read will continue in (BGEN)
+        long long admit_max_new = 0;
+        auto batch_on = [&] { for (const BSlot& b : bs) if (b.active) return true; return false; };
+        auto try_next_line = [&](std::string& out) -> bool {
+            std::lock_guard<std::mutex> lk(in_mu);
+            if (in_lines.empty()) return false;
+            out = std::move(in_lines.front());
+            in_lines.pop_front();
+            return true;
+        };
+        // the session a request just left behind (its prompt) -> slot b's sessions, on every stage
+        auto copy_to_slot = [&](int b, const std::vector<int32_t>& ids, std::string& e) -> bool {
+            const int64_t upto = (int64_t) ids.size();
+            for (size_t k = 0; k < bslot_ss.size(); ++k) {
+                strata::core::SessionState& from = k == 0 ? ss : stages[k - 1]->ss;
+                strata::core::SessionState& to = *bslot_ss[k][(size_t) b];
+                const strata::core::OnDevice on_k(k == 0 ? 0 : stages[k - 1]->dev);
+                if (cudaDeviceSynchronize() != cudaSuccess) { e = "batch admission: device sync failed"; return false; }
+                strata::core::ConversationCheckpoint ck;
+                ck.ids = ids;
+                if (!strata::core::conversation_checkpoint_save(ck, from, g, e) ||
+                    !strata::core::conversation_checkpoint_restore(ck, to, g, e))
+                    return false;
+                for (int64_t j = 0; j < from.qsa_alloc; ++j) {
+                    strata::core::ConversationKv img;
+                    if (!strata::core::conversation_kv_save(img, from.qsa_states[from.qsa_ord0 + j], g, upto, true, e))
+                        return false;
+                    if (cudaDeviceSynchronize() != cudaSuccess) { e = "batch admission: device sync failed"; return false; }
+                    if (!strata::core::conversation_kv_restore(img, to.qsa_states[to.qsa_ord0 + j], g, upto, true, e))
+                        return false;
+                }
+                if (cudaDeviceSynchronize() != cudaSuccess) { e = "batch admission: device sync failed"; return false; }
+            }
+            return true;
+        };
+        // one batch window over slots [0, highest active], inactive slots below it fed a pad token at position 0
+        auto batch_step = [&]() -> bool {
+            int S = 0;
+            for (int b = 0; b < (int) bs.size(); ++b) if (bs[(size_t) b].active) S = b + 1;
+            if (S == 0) return true;
+            int32_t tok[strata::kernels::kVerifyMaxT] = {}, outb[strata::kernels::kVerifyMaxT] = {};
+            int64_t pos[strata::kernels::kVerifyMaxT] = {};
+            for (int b = 0; b < S; ++b)
+                if (bs[(size_t) b].active) { tok[b] = bs[(size_t) b].x; pos[b] = bs[(size_t) b].p; }
+            strata::core::progress().busy.store(true);
+            drive.d.layers = 0;
+            drive.d.experts = 0;
+            drive.d.failed = false;
+            apply_pending(false);
+            if (bt_windows == 0) {
+                bt_start = Clock::now();
+                bt_wait0 = ver.ms_wait; bt_pool0 = ver.ms_pool;
+                bt_miss0 = drive.d.multi_misses; bt_hits0 = drive.d.cache_hits; bt_pcie0 = drive.d.pcie_experts;
+            }
+            const Clock::time_point w0 = Clock::now();
+            if (!ver.run_slots(S, tok, pos, win_pool_fn, win_pool_user, outb, err) || drive.d.failed) {
+                std::printf("ERR %s\n", drive.d.failed && drive.d.fail ? drive.d.fail : err.c_str());
+                return false;
+            }
+            const Clock::time_point w1 = Clock::now();
+            if (!ver.commit_slots(err)) {
+                std::printf("ERR %s\n", err.c_str());
+                return false;
+            }
+            const Clock::time_point w2 = Clock::now();
+            auto msd = [](Clock::time_point a0, Clock::time_point b0) { return std::chrono::duration<double, std::milli>(b0 - a0).count(); };
+            bt_run += msd(w0, w1);
+            bt_commit += msd(w1, w2);
+            ++bt_windows;
+            bt_rows += S;
+            for (int b = 0; b < S; ++b) {
+                BSlot& sl = bs[(size_t) b];
+                if (!sl.active) continue;
+                const int32_t y = outb[b];
+                std::printf("BT %d %d\n", b, (int) y);
+                ++sl.produced;
+                const bool eos = std::find(o.eos_ids.begin(), o.eos_ids.end(), (int64_t) y) != o.eos_ids.end();
+                const char* fin = eos ? "stop" : sl.stop ? "cancel" : sl.produced >= sl.max_new ? "length"
+                                : sl.p + 2 > o.max_context ? "length" : nullptr;
+                if (fin != nullptr) {
+                    const double ms = std::chrono::duration<double, std::milli>(Clock::now() - sl.t0).count();
+                    std::printf("BDONE %d %lld %s %.1f\n", b, (long long) sl.produced, fin, ms);
+                    sl.active = false;
+                } else {
+                    sl.x = y;
+                    sl.p += 1;
+                }
+            }
+            std::fflush(stdout);
+            bt_emit += msd(w2, Clock::now());
+            strata::core::progress().busy.store(false);
+            if (!batch_on() && bt_windows > 0) {
+                const double w = (double) bt_windows, wall = msd(bt_start, Clock::now());
+                const double L = (double) g.n_layers;
+                std::fprintf(stderr, "strata batch: %lld windows, avg %.2f rows, %.2f ms/window = run %.2f (CUDA0 GPU-reach "
+                                     "wait %.2f + CPU experts %.2f) + commit %.2f + emit %.2f; per layer-window: CPU experts "
+                                     "%.2f, VRAM hits %.2f, PCIe %.2f; %.1f rows/s over %.0f ms of wall time (admissions "
+                                     "included)\n",
+                             (long long) bt_windows, bt_rows / w, (bt_run + bt_commit + bt_emit) / w, bt_run / w,
+                             (ver.ms_wait - bt_wait0) / w, (ver.ms_pool - bt_pool0) / w, bt_commit / w, bt_emit / w,
+                             (drive.d.multi_misses - bt_miss0) / (w * L), (drive.d.cache_hits - bt_hits0) / (w * L),
+                             (drive.d.pcie_experts - bt_pcie0) / (w * L), 1000.0 * bt_rows / std::max(wall, 1e-9), wall);
+                for (size_t k = 0; k <= stages.size(); ++k) {
+                    const std::string pr = (k == 0 ? ver : stages[k - 1]->ver).profile_report();
+                    if (!pr.empty()) std::fprintf(stderr, "strata batch GPU stages, stage %zu (ms/window):%s\n", k + 1, pr.c_str());
+                }
+                bt_run = bt_commit = bt_emit = 0;
+                bt_windows = bt_rows = bt_tokens = 0;
+            }
+            return true;
+        };
+        // ---- --batch-groups: the slot groups pipelined through the stages (--batch-groups G > 1 with a layer split)
+        const int n_pipe = (int) stages.size() + 1;
+        const bool piped = o.batch > 0 && o.batch_groups > 1 && n_pipe > 1;
+        if (o.batch > 0 && o.batch_groups > 1 && (o.batch % o.batch_groups != 0 || !piped)) {
+            std::fprintf(stderr, "strata serve: --batch-groups needs a layer split and to divide --batch\n");
+            return 1;
+        }
+        const int GS = piped ? o.batch / o.batch_groups : o.batch;
+        struct PGroup {
+            bool inflight = false;
+            int stage = 0;                  ///< the stage it runs on or waits for
+            int32_t tok[strata::kernels::kVerifyMaxT] = {};
+            int64_t pos[strata::kernels::kVerifyMaxT] = {};
+            int64_t since = 0;              ///< tick it started waiting (fairness)
+        };
+        std::vector<PGroup> pg((size_t) (piped ? o.batch_groups : 0));
+        std::vector<int> stage_group((size_t) n_pipe, -1);
+        int64_t pipe_tick = 0, rr = 0;
+        auto stage_verifier = [&](int k) -> strata::core::Verifier& { return k == 0 ? ver : stages[(size_t) k - 1]->ver; };
+        auto pipe_inflight = [&] { for (const PGroup& x : pg) if (x.inflight) return true; return false; };
+        auto group_active = [&](int gi) {
+            for (int t = 0; t < GS; ++t) if (bs[(size_t) (gi * GS + t)].active) return true;
+            return false;
+        };
+        // one turn: serve every running stage, hand finished groups on, start what can start
+        auto pump = [&](bool may_start) -> bool {
+            ++pipe_tick;
+            if (bt_windows == 0 && !pipe_inflight()) {
+                bt_start = Clock::now();
+                bt_wait0 = ver.ms_wait; bt_pool0 = ver.ms_pool;
+                bt_miss0 = drive.d.multi_misses; bt_hits0 = drive.d.cache_hits; bt_pcie0 = drive.d.pcie_experts;
+            }
+            for (int k = 0; k < n_pipe; ++k) {
+                const int gi = stage_group[(size_t) k];
+                if (gi < 0) continue;
+                strata::core::Verifier& vk = stage_verifier(k);
+                const int r = vk.batch_poll(win_pool_fn, win_pool_user, err);
+                if (r < 0) { std::printf("ERR %s\n", err.c_str()); return false; }
+                if (r == 0) continue;
+                stage_group[(size_t) k] = -1;
+                PGroup& G = pg[(size_t) gi];
+                if (k + 1 < n_pipe) { G.stage = k + 1; G.since = pipe_tick; continue; }
+                // the last stage: the group's picks
+                const int32_t* outb = vk.batch_out();
+                for (int t = 0; t < GS; ++t) {
+                    BSlot& sl = bs[(size_t) (gi * GS + t)];
+                    if (!sl.active) continue;
+                    const int32_t y = outb[t];
+                    std::printf("BT %d %d\n", gi * GS + t, (int) y);
+                    ++sl.produced;
+                    ++bt_rows;
+                    const bool eos = std::find(o.eos_ids.begin(), o.eos_ids.end(), (int64_t) y) != o.eos_ids.end();
+                    const char* fin = eos ? "stop" : sl.stop ? "cancel" : sl.produced >= sl.max_new ? "length"
+                                    : sl.p + 2 > o.max_context ? "length" : nullptr;
+                    if (fin != nullptr) {
+                        const double ms = std::chrono::duration<double, std::milli>(Clock::now() - sl.t0).count();
+                        std::printf("BDONE %d %lld %s %.1f\n", gi * GS + t, (long long) sl.produced, fin, ms);
+                        sl.active = false;
+                    } else {
+                        sl.x = y;
+                        sl.p += 1;
+                    }
+                }
+                std::fflush(stdout);
+                ++bt_windows;
+                G.inflight = false;
+            }
+            // start waiting groups on free stages, the longest waiting first; a new step on stage 0 round-robin
+            for (int k = n_pipe - 1; k >= 0; --k) {
+                if (stage_group[(size_t) k] >= 0) continue;
+                int pick = -1;
+                if (k > 0) {
+                    for (int gi = 0; gi < (int) pg.size(); ++gi)
+                        if (pg[(size_t) gi].inflight && pg[(size_t) gi].stage == k &&
+                            (pick < 0 || pg[(size_t) gi].since < pg[(size_t) pick].since))
+                            pick = gi;
+                    if (pick < 0) continue;
+                } else {
+                    if (!may_start) continue;
+                    for (int j = 0; j < (int) pg.size() && pick < 0; ++j) {
+                        const int gi = (int) ((rr + j) % (int64_t) pg.size());
+                        if (!pg[(size_t) gi].inflight && group_active(gi)) pick = gi;
+                    }
+                    if (pick < 0) continue;
+                    rr = pick + 1;
+                    PGroup& G = pg[(size_t) pick];
+                    for (int t = 0; t < GS; ++t) {
+                        const BSlot& sl = bs[(size_t) (pick * GS + t)];
+                        G.tok[t] = sl.active ? sl.x : 0;
+                        G.pos[t] = sl.active ? sl.p : 0;
+                    }
+                    G.inflight = true;
+                    G.stage = 0;
+                    drive.d.failed = false;
+                    apply_pending(false);
+                }
+                PGroup& G = pg[(size_t) pick];
+                strata::core::progress().busy.store(true);
+                if (!stage_verifier(k).batch_launch(pick * GS, GS, G.tok, G.pos, err)) {
+                    std::printf("ERR %s\n", err.c_str());
+                    return false;
+                }
+                stage_group[(size_t) k] = pick;
+            }
+            if (drive.d.failed) { std::printf("ERR %s\n", drive.d.fail ? drive.d.fail : "the expert pool failed"); return false; }
+            if (!pipe_inflight() && !batch_on() && bt_windows > 0) {   // all idle: the timing line, as batch_step
+                const double w = (double) bt_windows, wall = std::chrono::duration<double, std::milli>(Clock::now() - bt_start).count();
+                std::fprintf(stderr, "strata batch (pipelined, %d groups of %d): %lld group-steps, %lld rows in %.0f ms = "
+                                     "%.1f rows/s (admissions included)\n", o.batch_groups, GS, (long long) bt_windows,
+                             (long long) bt_rows, wall, 1000.0 * bt_rows / std::max(wall, 1e-9));
+                for (int k = 0; k < n_pipe; ++k) {
+                    const std::string pr = stage_verifier(k).profile_report();
+                    if (!pr.empty()) std::fprintf(stderr, "strata batch GPU stages, stage %d (ms/window):%s\n", k + 1, pr.c_str());
+                }
+                bt_windows = bt_rows = 0;
+                strata::core::progress().busy.store(false);
+            }
+            return true;
+        };
+        auto pipe_drain = [&]() -> bool {
+            while (pipe_inflight())
+                if (!pump(false)) return false;
+            return true;
+        };
+        for (;;) {
+            if (batch_on() || (piped && pipe_inflight())) {
+                if (!try_next_line(line)) {
+                    if (!(piped ? pump(true) : batch_step())) return 1;
+                    continue;
+                }
+                // a request reads its prompt through every stage: the groups in flight finish first
+                if (piped && line.rfind("BSTOP ", 0) != 0 && !pipe_drain()) return 1;
+            } else if (!next_line(line)) {
+                break;
+            }
+            // --batch: BSTOP <slot> ends that slot at its next batch window; BGEN <slot> <max_new> ... reads the
+            // request's prompt and first token as a GEN 1, then continues it in that slot
+            if (line.rfind("BSTOP ", 0) == 0) {
+                const int b = std::atoi(line.c_str() + 6);
+                if (b >= 0 && b < (int) bs.size()) bs[(size_t) b].stop = true;
+                continue;
+            }
+            admit_slot = -1;
+            if (line.rfind("BGEN ", 0) == 0 || line.rfind("BGENI ", 0) == 0) {
+                const bool bimg = line.rfind("BGENI ", 0) == 0;
+                char* e1 = nullptr;
+                const long b = std::strtol(line.c_str() + (bimg ? 6 : 5), &e1, 10);
+                char* e2 = nullptr;
+                const long long mn = std::strtoll(e1, &e2, 10);
+                if (bs.empty() || b < 0 || b >= (long) bs.size() || bs[(size_t) b].active || mn < 1 || e2 == e1) {
+                    std::printf("ERR BGEN: no such free slot (--batch %d) or a bad max_new\n", o.batch);
+                    std::fflush(stdout);
+                    continue;
+                }
+                admit_slot = (int) b;
+                admit_max_new = mn;
+                line = std::string(bimg ? "GENI 1" : "GEN 1") + e2;
+            }
             // #477: every --expert-profile-save-every minutes, before the next request (at QUIT: after the loop)
             if (!heat.empty() && line != "QUIT" && o.expert_profile_save_min > 0 &&
                 Clock::now() - profile_saved_at >= std::chrono::duration<double>(o.expert_profile_save_min * 60.0))
@@ -5958,6 +6333,27 @@ int main(int argc, char** argv) {
                         (long long) (src.ram_reads() - ram0), (long long) (src.file_reads() - files0),
                         (double) (src.file_read_bytes() - file_bytes0) / 1e6, (long long) read_n);
             std::fflush(stdout);
+            if (admit_slot >= 0) {   // --batch: BADM <slot> <1 = continues in the batch windows | 0 = done>
+                bool cont = !cancelled && produced_n == 1 && admit_max_new > 1 && std::strcmp(finish, "length") == 0;
+                if (cont && !copy_to_slot(admit_slot, live, err)) {
+                    std::fprintf(stderr, "strata serve: batch admission failed: %s\n", err.c_str());
+                    cont = false;
+                }
+                if (cont) {
+                    ver.set_slot_sampling(admit_slot, req_sp);   // the request's own sampling, row by row
+                    BSlot& sl = bs[(size_t) admit_slot];
+                    sl = BSlot{};
+                    sl.active = true;
+                    sl.x = x;
+                    sl.p = p;
+                    sl.produced = 1;
+                    sl.max_new = admit_max_new;
+                    sl.t0 = Clock::now();
+                }
+                std::printf("BADM %d %d\n", admit_slot, cont ? 1 : 0);
+                std::fflush(stdout);
+                admit_slot = -1;
+            }
             if (drive.routing != nullptr) std::fflush(drive.routing);   // the routing trace survives a crash and is watchable mid-session
             // "12288 of 98179" when cancelled mid-read (#471), the rate from what was read
             char read_txt[64];

tools/batch_test.py

@@ -0,0 +1,153 @@
+#!/usr/bin/env python3
+"""Batching test: the same prompts decoded alone (GEN, the usual path with drafts) and together in the batch
+windows (BGEN), greedy.  Every slot's tokens must equal its solo tokens; prints the aggregate decode rate.
+
+  python3 tools/batch_test.py --exe engine/strata --config strata-<model>.json --batch 4 --n 4 \
+      --extra "--layer-split 12,24,36 --trim-stage-weights --pcie-frac 0 --adapt-every 1000000"
+"""
+import argparse, json, os, subprocess, sys, threading, time
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import strata_tokenizer as ST  # noqa: E402
+
+QUESTIONS = [
+    "Explique en detail le fonctionnement d'un B-tree.",
+    "Write a Python function that merges overlapping intervals, then explain it.",
+    "Compare TCP and QUIC: handshake, congestion control, multiplexing.",
+    "Raconte l'histoire du calcul de pi, d'Archimede a nos jours.",
+    "What are the main causes of the French Revolution?",
+    "Explain how a transformer's attention works, step by step.",
+    "Donne une recette de tarte aux pommes, etape par etape.",
+    "Describe the life cycle of a star like the Sun.",
+]
+
+
+def tokenizer(path):
+    t = Path(path)
+    vocab = json.loads((t / "vocab.json").read_text(encoding="utf-8"))
+    tokens = [None] * len(vocab)
+    for s, i in vocab.items():
+        tokens[i] = s
+    merges = (t / "merges.txt").read_text(encoding="utf-8").split("\n")
+    types = json.loads((t / "token_type.json").read_text())
+    return ST.Tokenizer(tokens, merges, types)
+
+
+class Engine:
+    def __init__(self, exe, cfg, batch, extra_env, extra_args):
+        args = list(cfg["args"])
+        if len(cfg.get("gpu") or []) > 1:
+            args += ["--layer-split", str(cfg.get("layer_split") or "auto")]
+        if batch:
+            args += ["--batch", str(batch)]
+        args += extra_args
+        env = dict(os.environ, **extra_env)
+        env["LD_LIBRARY_PATH"] = ":".join(cfg.get("lib_dirs", []) + [env.get("LD_LIBRARY_PATH", "")])
+        self.log = open("/tmp/batch_test_engine.log", "w")
+        self.p = subprocess.Popen([exe, "--serve", *args], cwd=cfg.get("cwd"), stdin=subprocess.PIPE,
+                                  stdout=subprocess.PIPE, stderr=self.log, text=True, bufsize=1, env=env)
+        for line in self.p.stdout:
+            if line.startswith("READY"):
+                return
+        raise SystemExit("the engine ended before READY - see /tmp/batch_test_engine.log")
+
+    def send(self, line):
+        self.p.stdin.write(line + "\n")
+        self.p.stdin.flush()
+
+    def lines(self):
+        for line in self.p.stdout:
+            yield line.rstrip("\n")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--exe", required=True)
+    ap.add_argument("--config", required=True)
+    ap.add_argument("--batch", type=int, default=4)
+    ap.add_argument("--n", type=int, default=4, help="prompts (<= --batch)")
+    ap.add_argument("--max-new", type=int, default=64)
+    ap.add_argument("--skip-solo", action="store_true")
+    ap.add_argument("--extra", default="", help='more engine arguments in one string, e.g. "--adapt-every 1000000"')
+    a = ap.parse_args()
+    cfg = json.loads(Path(a.config).read_text())
+    tok = tokenizer(cfg["tokenizer"])
+    prompts = []
+    for q in QUESTIONS[: a.n]:
+        text = f"<|im_start|>user\n{q}<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
+        prompts.append(tok.encode(text, parse_special=True))
+    eng = Engine(a.exe, cfg, a.batch, {"STRATA_IQ_MT_MIN": "1", "STRATA_DECODE_TIMING": "1", **({"STRATA_VERIFY_PROFILE": "1"} if os.environ.get("PROF") else {})}, a.extra.split())
+    out = eng.lines()
+
+    solo = []
+    if not a.skip_solo:
+        for i, ids in enumerate(prompts):
+            eng.send(f"GEN {a.max_new} " + ",".join(map(str, ids)))
+            got, t0 = [], time.time()
+            for line in out:
+                if line.startswith("T "):
+                    got.append(int(line.split()[1]))
+                elif line.startswith("DONE") or line.startswith("ERR"):
+                    print(f"solo {i}: {len(got)} tokens in {time.time() - t0:.1f}s  {line[:60]}", flush=True)
+                    break
+            solo.append(got)
+
+    # batch: admit every prompt, then the slots decode together
+    got = {i: [] for i in range(len(prompts))}
+    done = {}
+    t_admit = time.time()
+    first_bt = None
+    for i, ids in enumerate(prompts):
+        eng.send(f"BGEN {i} {a.max_new} " + ",".join(map(str, ids)))
+        for line in out:
+            if line.startswith("T "):
+                got[i].append(int(line.split()[1]))
+            elif line.startswith("BT "):
+                _, s, y = line.split()
+                got[int(s)].append(int(y))
+                first_bt = first_bt or time.time()
+            elif line.startswith("BDONE "):
+                done[int(line.split()[1])] = line
+            elif line.startswith("ERR"):
+                print("batch:", line)
+                return 1
+            elif line.startswith("BADM "):
+                if line.split()[2] == "0":
+                    done[i] = line
+                break
+    t_admitted = time.time()
+    for line in out:
+        if line.startswith("BT "):
+            _, s, y = line.split()
+            got[int(s)].append(int(y))
+            first_bt = first_bt or time.time()
+        elif line.startswith("BDONE "):
+            done[int(line.split()[1])] = line
+            if len(done) == len(prompts):
+                break
+        elif line.startswith("ERR"):
+            print("batch:", line)
+            return 1
+    t_end = time.time()
+    total = sum(len(v) for v in got.values())
+    print(f"batch: {len(prompts)} slots, {total} tokens; admissions {t_admitted - t_admit:.1f}s; "
+          f"then {t_end - t_admitted:.1f}s -> aggregate {total / max(t_end - t_admit, 1e-9):.1f} tok/s overall, "
+          f"{sum(len(v) for v in got.values()) / max(t_end - (first_bt or t_admit), 1e-9):.1f} tok/s from the first batch token",
+          flush=True)
+    ok = True
+    for i in range(len(prompts)):
+        b = got[i]
+        if solo:
+            same = b == solo[i]
+            ok &= same
+            first_diff = next((k for k in range(min(len(b), len(solo[i]))) if b[k] != solo[i][k]), None)
+            print(f"slot {i}: {len(b)} tokens, solo {len(solo[i])}: {'IDENTICAL' if same else f'DIFFERS at {first_diff}'}")
+        print("   ", repr(tok.decode(b)[:160]))
+    eng.send("QUIT")
+    eng.p.wait(timeout=180)   # the next run needs the GPUs back
+    return 0 if ok else 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())