--parallel slots honour the request's repeat / frequency / presence penalties and logit bias (CUDA, HIP): each penalised slot's reply tail is staged before its window (set_slot_history), the verifier builds the per-row history (slot_penalty_rows: the tail plus the slot's tokens in the window up to the row) and draws the row with sample_tokens; the slot's own logit bias array (set_slot_logit_bias); a logit_bias request is no longer refused with --batch; plain greedy slots keep the window's argmax. CPU test slot_history_test

CMakeLists.txt

@@ -1946,6 +1946,10 @@ if(STRATA_BUILD_TESTS OR STRATA_BUILD_CONVERSATION_TESTS)
   add_executable(penalty_window_test src/kernels/penalty_window_test.cpp)
   target_include_directories(penalty_window_test PRIVATE include)
   add_test(NAME penalty_window_test COMMAND penalty_window_test)
+  # the penalty-history rows of a batch window, per slot (header-only, CPU)
+  add_executable(slot_history_test src/kernels/slot_history_test.cpp)
+  target_include_directories(slot_history_test PRIVATE include)
+  add_test(NAME slot_history_test COMMAND slot_history_test)
   # probabilistic draft acceptance (STRATA_SPEC_PROB): rejection sampling preserves p - chi-square over 2M trials per case (CPU)
   add_executable(spec_prob_test src/core/spec_prob_test.cpp)
   target_include_directories(spec_prob_test PRIVATE include)

include/strata/core/verify.hpp

@@ -281,9 +281,20 @@ public:
     /// 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 (slot >= 0 && slot < (int) slot_sp_.size()) {
+            slot_sp_[(size_t) slot] = sp;
+            slot_sp_[(size_t) slot].logit_bias = slot < (int) slot_bias_dev_.size() ? slot_bias_dev_[(size_t) slot] : nullptr;   // the slot's own bias array (set_slot_logit_bias)
+        }
         if (next_) next_->set_slot_sampling(slot, sp);
     }
+    /// A slot's logit bias (n_vocab floats; empty = none), added before the penalties and filters of its row, as a solo request's.  Reaches the last stage.
+    bool set_slot_logit_bias(int slot, const std::vector<float>& bias, std::string& err);
+    /// The slot's reply tail for its penalty window: the `n` tokens its session consumed since the reply began (the last penalty_last_n are used), the
+    /// fed-back head not included.  Set before every batch window that holds the slot while it has a penalty; reaches the last stage.
+    void set_slot_history(int slot, const int32_t* tail, int64_t n) {
+        if (slot >= 0 && slot < (int) slot_tail_.size()) slot_tail_[(size_t) slot].assign(tail, tail + (n > 0 ? n : 0));
+        if (next_) next_->set_slot_history(slot, tail, n);
+    }
     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
@@ -447,6 +458,10 @@ private:
     std::chrono::steady_clock::time_point ext_checked_{};   ///< batch_poll's last check with external serving
     int32_t b_out_[strata::kernels::kWindowMaxT] = {};
     std::vector<strata::kernels::SamplerParams> slot_sp_;
+    std::vector<float*> slot_bias_dev_;                  ///< per slot: its logit bias on the device (null = none)
+    std::vector<std::vector<int32_t>> slot_tail_;        ///< per slot: its reply tail for the penalty window (set_slot_history)
+    int32_t* slot_hist_d_ = nullptr;                     ///< the penalty-history rows of the current batch window (device)
+    size_t slot_hist_cap_ = 0;
     bool sample_rows(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;

include/strata/kernels/sampler.hpp

@@ -121,4 +121,29 @@ inline void penalty_rows(const int32_t* tail, int64_t n_tail, const int32_t* win
     }
 }
 
+/// The penalty-history rows of a BATCH window, on the host.  Row t of the window feeds `tokens[t]` into slot `rows[t]`; its row of `out` (h slots, most
+/// recent LAST, -1 in unused front slots) is the last `h` tokens of that slot's reply tail (`tails[slot]`, `n_tails[slot]` tokens: what the slot's
+/// session consumed since the reply began, the fed-back head not yet included) followed by the tokens fed so far by THIS window's rows of the same slot
+/// up to and including row t (a slot may hold several rows: MTP drafts).  Exactly what `penalty_rows` gives a solo window, per slot.
+inline void slot_penalty_rows(const int* rows, const int32_t* tokens, int S, const int32_t* const* tails, const int64_t* n_tails, int h, int32_t* out) {
+    for (int t = 0; t < S; ++t) {
+        int32_t* row = out + (size_t) t * (size_t) h;
+        const int slot = rows[t];
+        const int32_t* tail = tails[slot];
+        const int64_t n_tail = tail != nullptr ? n_tails[slot] : 0;
+        int64_t in_window = 0;                                     // this window's rows of the slot up to t
+        for (int u = 0; u <= t; ++u) in_window += rows[u] == slot;
+        const int64_t avail = n_tail + in_window;
+        const int take = (int) std::min<int64_t>(h, avail);
+        std::fill(row, row + (h - take), -1);
+        for (int j = 0; j < take; ++j) {
+            const int64_t i = avail - take + j;                     // index into (tail, the slot's window tokens)
+            if (i < n_tail) { row[h - take + j] = tail[i]; continue; }
+            int64_t k = i - n_tail, seen = 0;
+            for (int u = 0; u <= t; ++u)
+                if (rows[u] == slot) { if (seen == k) { row[h - take + j] = tokens[u]; break; } ++seen; }
+        }
+    }
+}
+
 }  // namespace strata::kernels

src/core/verify.cpp

@@ -733,6 +733,8 @@ Verifier::~Verifier() {
     if (tap_scratch_) cudaFree(tap_scratch_);
     if (h_tap_) cudaFreeHost(h_tap_);
     if (logit_bias_device_) cudaFree(logit_bias_device_);
+    for (float* b : slot_bias_dev_) if (b) cudaFree(b);
+    if (slot_hist_d_) cudaFree(slot_hist_d_);
     if (cs_ && cs_ != ext_stream_) cudaStreamDestroy(cs_);   // set_stream: the stage's stream, shared, not ours
     if (sh_cs_) cudaStreamDestroy(sh_cs_);
     if (copy_) { cudaStreamSynchronize(copy_); cudaStreamDestroy(copy_); }
@@ -3834,6 +3836,8 @@ bool Verifier::init_slots(const std::vector<SessionState*>& slots, std::string&
     // the Metal window keeps each slot's running state itself (metal/src/engine/window.cpp): none of the CUDA batch buffers
     slots_ = slots;
     slot_sp_.assign(slots.size(), sampling_);
+    slot_bias_dev_.assign(slots.size(), nullptr);
+    slot_tail_.assign(slots.size(), {});
     if (metal_ == nullptr || !strata::metal_engine::window_init_slots(metal_, slots, err)) {
         if (err.empty()) err = "verify: the Metal window is not up";
         return false;
@@ -3859,6 +3863,8 @@ bool Verifier::init_slots(const std::vector<SessionState*>& slots, std::string&
     tail_snap_b_ = (float*) ((uint8_t*) d + a);
     slots_ = slots;
     slot_sp_.assign(slots.size(), sampling_);   // greedy until set_slot_sampling
+    slot_bias_dev_.assign(slots.size(), nullptr);
+    slot_tail_.assign(slots.size(), {});
     std::fprintf(stderr, "strata verify: batch windows of up to %lld sequences (layers [%lld, %lld))%s\n", (long long) S,
                  (long long) lb_, (long long) le_, g_batch_self_commit ? ", one-row windows keep their rows" : "");
     return true;
@@ -4706,14 +4712,59 @@ bool Verifier::commit_slot_prefixes(const int* keep, std::string& err) {
     return next_ == nullptr || next_->commit_slot_prefixes(keep, err);
 }
 
+bool Verifier::set_slot_logit_bias(int slot, const std::vector<float>& bias, std::string& err) {
+    if (next_ && !next_->set_slot_logit_bias(slot, bias, err)) return false;
+    if (le_ < g_->n_layers) return true;
+    if (slot < 0 || slot >= (int) slot_bias_dev_.size()) return true;
+    const OnDevice on_device(device_);
+    if (bias.empty()) { slot_bias_dev_[(size_t) slot] = nullptr; slot_sp_[(size_t) slot].logit_bias = nullptr; return true; }
+    if (bias.size() != (size_t) n_vocab_) { err = "logit_bias vocabulary size mismatch"; return false; }
+    const size_t bytes = bias.size() * sizeof(float);
+    if (slot_bias_dev_[(size_t) slot] == nullptr) {
+        float* d = nullptr;
+        if (cudaMalloc((void**) &d, bytes) != cudaSuccess) { err = "cannot allocate the slot's logit_bias"; return false; }
+        slot_bias_dev_[(size_t) slot] = d;
+    }
+    if (cudaMemcpy(slot_bias_dev_[(size_t) slot], bias.data(), bytes, cudaMemcpyHostToDevice) != cudaSuccess) { err = "cannot upload the slot's logit_bias"; return false; }
+    slot_sp_[(size_t) slot].logit_bias = slot_bias_dev_[(size_t) slot];
+    return true;
+}
+
 bool Verifier::sample_rows(int S, std::string& err) {
     bool any = false;
+    // the slots' penalty-history rows of this window (host), built once for the rows that need one
+    constexpr int kSlotHistMax = 4096;   // serve's kPenaltyWindowCap
+    std::vector<int32_t> hrows;
+    std::vector<const int32_t*> tails(slots_.size(), nullptr);
+    std::vector<int64_t> n_tails(slots_.size(), 0);
+    for (size_t b = 0; b < slots_.size() && b < slot_tail_.size(); ++b) { tails[b] = slot_tail_[b].data(); n_tails[b] = (int64_t) slot_tail_[b].size(); }
     for (int t = 0; t < S; ++t) {
         strata::kernels::SamplerParams sp = slot_sp_[(size_t) last_rows_[t]];
-        if (sp.greedy || sp.temperature <= 0.0f) continue;
+        const bool pen = sp.penalty_last_n > 0 && (sp.penalty_repeat != 1.0f || sp.penalty_freq != 0.0f || sp.penalty_present != 0.0f);
+        const bool bias = sp.logit_bias != nullptr;
+        if ((sp.greedy || sp.temperature <= 0.0f) && !pen && !bias) continue;   // a plain greedy slot keeps the window's argmax token untouched
         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,
+        const int32_t* hist = nullptr;
+        int hlen = 0;
+        if (pen) {
+            hlen = std::min(sp.penalty_last_n, kSlotHistMax);
+            hrows.assign((size_t) S * (size_t) hlen, -1);
+            strata::kernels::slot_penalty_rows(last_rows_, last_tokens_, S, tails.data(), n_tails.data(), hlen, hrows.data());
+            const size_t need = (size_t) S * (size_t) hlen;
+            const OnDevice on_hist(device_);
+            if (slot_hist_cap_ < need) {
+                if (slot_hist_d_ != nullptr) cudaFree(slot_hist_d_);
+                slot_hist_d_ = nullptr;
+                if (cudaMalloc((void**) &slot_hist_d_, need * sizeof(int32_t)) != cudaSuccess) { err = "verify batch: the penalty-history allocation failed"; return false; }
+                slot_hist_cap_ = need;
+            }
+            if (cudaMemcpyAsync(slot_hist_d_ + (size_t) t * (size_t) hlen, hrows.data() + (size_t) t * (size_t) hlen, (size_t) hlen * sizeof(int32_t),
+                                cudaMemcpyHostToDevice, cs_) != cudaSuccess) { err = "verify batch: the penalty-history copy failed"; return false; }
+            hist = slot_hist_d_ + (size_t) t * (size_t) hlen;
+        } else {
+            sp.penalty_last_n = 0;
+        }
+        strata::kernels::sample_tokens(head_logits_ + (size_t) t * (size_t) n_vocab_, 1, (int) n_vocab_, hist, hlen, sp,
                                        m_out_ + t, cs_);
         any = true;
     }

src/kernels/slot_history_test.cpp

@@ -0,0 +1,89 @@
+// Penalty-history rows of a BATCH window, per slot (strata::kernels::slot_penalty_rows).  Header-only, CPU, no model.
+//
+// Row t of a batch window feeds tokens[t] into slot rows[t]; its history is the slot's reply tail followed by that slot's tokens in this window up to
+// and including row t - what penalty_rows gives a solo window of that slot, whatever the other slots do.
+#include "strata/kernels/sampler.hpp"
+
+#include <cstdio>
+#include <algorithm>
+#include <cstdlib>
+#include <random>
+#include <vector>
+
+using strata::kernels::penalty_rows;
+using strata::kernels::slot_penalty_rows;
+
+namespace {
+int checks = 0;
+void check(bool ok, const char* label) {
+    ++checks;
+    if (!ok) { std::fprintf(stderr, "FAIL: %s\n", label); std::exit(1); }
+}
+}  // namespace
+
+int main() {
+    std::mt19937 rng(7);
+    constexpr int kSlots = 6, kH = 16;
+    for (int trial = 0; trial < 400; ++trial) {
+        // per slot: a reply tail of 0..40 tokens
+        std::vector<std::vector<int32_t>> tail(kSlots);
+        std::vector<const int32_t*> tails(kSlots, nullptr);
+        std::vector<int64_t> n_tails(kSlots, 0);
+        for (int b = 0; b < kSlots; ++b) {
+            const int n = (int) (rng() % 41);
+            for (int i = 0; i < n; ++i) tail[b].push_back((int32_t) (rng() % 500));
+            tails[b] = tail[b].empty() ? nullptr : tail[b].data();
+            n_tails[b] = (int64_t) tail[b].size();
+        }
+        // a window: 1..8 rows; each slot at most once, or as a contiguous group of 1..3 rows (MTP drafts)
+        std::vector<int> rows; std::vector<int32_t> toks;
+        std::vector<int> order = {0, 1, 2, 3, 4, 5};
+        std::shuffle(order.begin(), order.end(), rng);
+        for (int b : order) {
+            if (rows.size() >= 8) break;
+            if (rng() % 3 == 0) continue;
+            const int g = 1 + (int) (rng() % 3);
+            for (int j = 0; j < g && rows.size() < 8; ++j) { rows.push_back(b); toks.push_back((int32_t) (rng() % 500)); }
+        }
+        if (rows.empty()) { rows.push_back(2); toks.push_back(9); }
+        const int S = (int) rows.size();
+        std::vector<int32_t> out((size_t) S * kH, 12345);
+        slot_penalty_rows(rows.data(), toks.data(), S, tails.data(), n_tails.data(), kH, out.data());
+        // the definition: per slot, penalty_rows over (its tail, its rows of this window)
+        for (int b = 0; b < kSlots; ++b) {
+            std::vector<int> idx;
+            std::vector<int32_t> win;
+            for (int t = 0; t < S; ++t) if (rows[t] == b) { idx.push_back(t); win.push_back(toks[t]); }
+            if (idx.empty()) continue;
+            std::vector<int32_t> ref(idx.size() * kH);
+            penalty_rows(tail[b].data(), (int64_t) tail[b].size(), win.data(), (int) idx.size(), kH, ref.data());
+            for (size_t k = 0; k < idx.size(); ++k)
+                for (int j = 0; j < kH; ++j)
+                    check(out[(size_t) idx[k] * kH + j] == ref[k * kH + j], "slot row equals penalty_rows of the slot alone");
+        }
+    }
+    // ---- fixed cases: an empty tail, the fed-back head counted, other slots' tokens never in a row
+    {
+        const int rows[3] = {0, 1, 1};
+        const int32_t toks[3] = {10, 20, 21};
+        const int32_t t0[2] = {1, 2};
+        const int32_t t1[3] = {7, 8, 9};
+        const int32_t* tails[2] = {t0, t1};
+        const int64_t n_tails[2] = {2, 3};
+        std::vector<int32_t> out(3 * 4, 0);
+        slot_penalty_rows(rows, toks, 3, tails, n_tails, 4, out.data());
+        const int32_t e[12] = {-1, 1, 2, 10,    7, 8, 9, 20,    8, 9, 20, 21};
+        for (int i = 0; i < 12; ++i) check(out[i] == e[i], "fixed case");
+    }
+    {   // a slot with no tail (no penalty history): its row holds the fed token only
+        const int rows[1] = {0};
+        const int32_t toks[1] = {42};
+        const int32_t* tails[1] = {nullptr};
+        const int64_t n_tails[1] = {0};
+        int32_t out[4] = {0, 0, 0, 0};
+        slot_penalty_rows(rows, toks, 1, tails, n_tails, 4, out);
+        check(out[0] == -1 && out[1] == -1 && out[2] == -1 && out[3] == 42, "empty tail");
+    }
+    std::printf("slot_history_test: %d checks passed\n", checks);
+    return 0;
+}

src/program/generate.cpp

@@ -13032,6 +13032,8 @@ int main(int argc, char** argv) {
             // fed).  Kept when the request ends (`cached`), so the next turn of that conversation continues from it
             // instead of reading its history again (see `slot_source` in the request path).
             std::vector<int32_t> ids;
+            int64_t reply0 = 0;            ///< the index in `ids` where its penalty window starts (the last prompt token; 0 with STRATA_PENALTY_PROMPT=1)
+            int pen_n = 0;                 ///< its request's penalty window (penalty_last_n) when it has a repeat / frequency / presence penalty, else 0
             bool cached = false;           ///< idle, and its sessions still hold `ids`
             bool cvec = true;              ///< the control vector setting `ids` were read with
             bool img = false;              ///< the conversation has pictures (never reused from the slot)
@@ -13372,6 +13374,12 @@ int main(int argc, char** argv) {
                 if (pc.on()) pc.begin(pc_ex(), (int) g.n_layers);
                 bt_a_rounds0 = a_rounds; bt_a_swapped0 = a_swapped;
             }
+            for (int t = 0; t < S; ++t) {   // each penalised slot's reply tail for its penalty window (the verifier builds the rows from it and this window's tokens)
+                const BSlot& ps = bs[(size_t) rows[t]];
+                if (ps.pen_n <= 0 || (t > 0 && rows[t - 1] == rows[t])) continue;
+                const int64_t from = std::max<int64_t>(ps.reply0, (int64_t) ps.ids.size() - ps.pen_n);
+                ver.set_slot_history(rows[t], ps.ids.data() + from, (int64_t) ps.ids.size() - from);
+            }
             const Clock::time_point w0 = Clock::now();
             if (!ver.run_slot_rows(rows, 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());
@@ -13939,6 +13947,16 @@ int main(int argc, char** argv) {
                 PGroup& G = pg[(size_t) pick];
                 strata::core::progress().busy.store(true);
                 if (batch_telemetry_enabled) stage_in_flight.launch(k, Clock::now());
+                if (k == 0) {   // the penalised slots' reply tails for the group's rows (the last stage draws them)
+                    int grows[strata::kernels::kVerifyMaxT] = {};
+                    for (int t = 0; t < G.S && t < (int) strata::kernels::kVerifyMaxT; ++t) grows[t] = pmtp ? G.rows[t] : pick * GS + t;
+                    for (int t = 0; t < G.S && t < (int) strata::kernels::kVerifyMaxT; ++t) {
+                        const BSlot& ps = bs[(size_t) grows[t]];
+                        if (ps.pen_n <= 0 || (t > 0 && grows[t - 1] == grows[t])) continue;
+                        const int64_t from = std::max<int64_t>(ps.reply0, (int64_t) ps.ids.size() - ps.pen_n);
+                        ver.set_slot_history(grows[t], ps.ids.data() + from, (int64_t) ps.ids.size() - from);
+                    }
+                }
                 const auto p4_l0 = p4.now();
                 if (!(pmtp ? stage_verifier(k).batch_launch_rows(G.rows, G.S, G.hbase, G.tok, G.pos, err)
                            : stage_verifier(k).batch_launch(pick * GS, G.S, G.tok, G.pos, err))) {
@@ -14489,9 +14507,8 @@ int main(int argc, char** argv) {
             bool req_df2 = use_df2;   // --draft auto: this request's drafter, decided once the prompt is parsed (below)
             std::vector<float> req_bias;
             std::string bias_error;
-            if (req_bias_present && (o.batch > 0 ||
-                !strata::core::parse_logit_bias(req_bias_text, (int) n_vocab, req_bias, bias_error))) {
-                std::printf("ERR logit_bias: %s\n", o.batch > 0 ? "continuous batching is not supported" : bias_error.c_str());
+            if (req_bias_present && !strata::core::parse_logit_bias(req_bias_text, (int) n_vocab, req_bias, bias_error)) {
+                std::printf("ERR logit_bias: %s\n", bias_error.c_str());
                 continue;
             }
             std::string emb_path;
@@ -18155,6 +18172,10 @@ int main(int argc, char** argv) {
                 const auto tc1 = Clock::now();
                 if (cont) {
                     ver.set_slot_sampling(admit_slot, req_sp);   // the request's own sampling, row by row
+                    if (!ver.set_slot_logit_bias(admit_slot, req_bias, err)) {   // and its own logit bias
+                        std::printf("ERR %s\n", err.c_str());
+                        return 1;
+                    }
                     BSlot& sl = bs[(size_t) admit_slot];
                     if (ckpt_pool.on())
                         for (ConvCheckpoint& c : sl.checks) ckpt_pool.give(std::move(c));
@@ -18167,6 +18188,11 @@ int main(int argc, char** argv) {
                     sl.max_new = admit_max_new;
                     sl.t0 = Clock::now();
                     sl.ids = live;
+                    {   // the penalty window of the slot's decode: the reply, from the last prompt token (the first window fed it), as the solo loop counts it
+                        const bool slot_pen = req_sp.penalty_last_n > 0 && (req_sp.penalty_repeat != 1.0f || req_sp.penalty_freq != 0.0f || req_sp.penalty_present != 0.0f);
+                        sl.pen_n = slot_pen ? std::min(req_sp.penalty_last_n, kPenaltyWindowCap) : 0;
+                        sl.reply0 = (int64_t) strata::kernels::penalty_window_from(live.size() > 0 ? live.size() - 1 : 0, pen_prompt);
+                    }
                     if (batch_mtp) {
                         int nd = 1;   // (STRATA_PIPED_MTP: up to its proposals per window, as --spec-min-p allows)
                         if (!slot_mtp[(size_t) admit_slot]->draft_first(1, ver.final_R_all(), sl.x,