feat(qwen35): serve image requests in the concurrent batch

docs/image-input.md

@@ -111,9 +111,14 @@ plain autoregressive decoding and bypass the token-keyed prefix, disk and
 agent-turn caches and prompt compression: tokens alone do not identify an
 image. Text requests on the same server keep speculative decoding and caching.
 
-Layer or tensor splitting across GPUs, remote target shards, concurrent
-sequence scheduling (`--max-concurrency`) and upstream forwarding do not
-support images. `/props` reports the effective capability in
+Layer or tensor splitting across GPUs, remote target shards and upstream
+forwarding do not support images. Qwen3.5 / Qwen3.8 serve images with
+concurrent sequence scheduling (`--paged-attention --max-concurrency N`): each
+image request is encoded when it is admitted and then prefills and decodes in
+the shared batch like text, with the drafter. On one R9700, four concurrent
+256-token image answers finish in 6.9 s (149 tok/s in total) against 13.3 s
+(77 tok/s) one at a time. DeepSeek V4 image requests still need one request
+at a time. `/props` reports the effective capability in
 `capabilities.image_input_supported` after backend initialization.
 
 ## Qwen3.5 / Qwen3.8

server/src/common/concurrency/seq_engine.h

@@ -54,6 +54,7 @@
 #include <string>
 #include <vector>
 
+#include "common/image_prompt.h"
 #include "common/sampler.h"
 #include "prefix_store.h"
 
@@ -181,6 +182,22 @@ class SeqEngine {
                               const std::vector<int32_t> & prompt,
                               const SamplerCfg & sampler) = 0;
 
+    // Image requests. The engine keeps the payload with the slot for as long
+    // as the slot lives, so a re-prefill after eviction sees the images again.
+    // Engines without image support refuse; the server only routes images to
+    // an engine that reports supports_images().
+    virtual bool supports_images() const { return false; }
+    virtual AdmitResult admit_images(uint64_t request_id,
+                                     const std::vector<int32_t> & prompt,
+                                     const SamplerCfg & sampler,
+                                     const ImagePromptHandle & images) {
+        (void) request_id; (void) prompt; (void) sampler; (void) images;
+        AdmitResult result;
+        result.status = AdmitResult::Status::failed;
+        result.error = "this engine does not serve image requests";
+        return result;
+    }
+
     // Optional prefix-checkpoint admission. Unsupported engines remain on
     // cold admission and never receive a plan from the scheduler.
     virtual bool supports_prefix_store() const { return false; }

server/src/common/feature_gate.cpp

@@ -51,9 +51,9 @@ std::string check_feature_compatibility(
     if (args.mmproj_path.has_value()) {
         if ((arch != "deepseek4" && arch != "qwen35") || args.device.is_layer_split() ||
             args.device.is_tensor_parallel() || args.remote_target_shard.enabled() ||
-            args.max_concurrency != 1) {
-            return "--mmproj requires a local single-request DeepSeek4 or Qwen3.5 backend "
-                   "that is not split across GPUs by layer or tensor";
+            (arch == "deepseek4" && args.max_concurrency != 1)) {
+            return "--mmproj requires a local DeepSeek4 (one request at a time) or Qwen3.5 "
+                   "backend that is not split across GPUs by layer or tensor";
         }
         if (arch == "deepseek4" && target_backend != PlacementBackend::Hip) {
             return "--mmproj with DeepSeek4 requires a HIP backend";

server/src/qwen35/concurrency/qwen35_seq_engine.cpp

@@ -379,6 +379,7 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit(
         const SamplerCfg & sampler) {
     AdmitResult result = slots_.admit(request_id, prompt, sampler);
     if (result.status == AdmitResult::Status::admitted) {
+        clear_slot_images(result.slot);
         reset_recurrent_slot(b_.cache_, result.slot);
         if (result.slot >= 0 &&
             result.slot < static_cast<int>(slot_draft_kv_.size()) &&
@@ -389,6 +390,46 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit(
     return result;
 }
 
+bool Qwen35SeqEngine::supports_images() const {
+    return b_.supports_images();
+}
+
+SeqEngine::AdmitResult Qwen35SeqEngine::admit_images(
+        uint64_t request_id,
+        const std::vector<int32_t> & prompt,
+        const SamplerCfg & sampler,
+        const ImagePromptHandle & images) {
+    AdmitResult refused;
+    refused.status = AdmitResult::Status::failed;
+    const auto * payload = dynamic_cast<const Qwen35ImagePrompt *>(images.get());
+    if (!payload || payload->owner != &b_ || !payload->matches(prompt)) {
+        refused.error = "image binding does not match this prompt";
+        return refused;
+    }
+    // Claim the slot first: a busy pool defers the request and retries it,
+    // and encoding before that would rerun the tower on every retry. The
+    // tower runs on this (the scheduler) thread, so no step sees the slot
+    // before its images are in place.
+    AdmitResult result = admit(request_id, prompt, sampler);
+    if (result.status != AdmitResult::Status::admitted) return result;
+    Qwen35ImageRows rows;
+    std::string error = "image encoding failed";
+    if (!b_.encode_images(*payload, rows, error)) {
+        retire(result.slot);
+        refused.error = error;
+        return refused;
+    }
+    if (slot_images_.size() < static_cast<size_t>(slots_.slot_count())) {
+        slot_images_.resize(static_cast<size_t>(slots_.slot_count()));
+    }
+    SlotImages & state = slot_images_[static_cast<size_t>(result.slot)];
+    state.payload = images;
+    state.rows = std::move(rows);
+    state.rows.prompt = payload;
+    state.rope_delta = payload->positions.next - static_cast<int>(prompt.size());
+    return result;
+}
+
 size_t Qwen35SeqEngine::estimate_prefix_store_bytes(int tokens) const {
     return estimate_paged_target_cache_snapshot_bytes(b_.cache_, tokens);
 }
@@ -425,6 +466,7 @@ SeqEngine::AdmitResult Qwen35SeqEngine::admit_with_prefix(
         const PrefixStorePlan & plan) {
     AdmitResult result = slots_.admit(request_id, prompt, sampler);
     if (result.status != AdmitResult::Status::admitted) return result;
+    clear_slot_images(result.slot);
 
     const int slot = result.slot;
     slots_.slot(slot).pending_capture = {};
@@ -652,6 +694,13 @@ Qwen35SeqEngine::PrefillStage Qwen35SeqEngine::stage_prefill_chunk(
                      "prefill embedding failed");
         return PrefillStage{};
     }
+    if (slot < static_cast<int>(slot_images_.size()) &&
+        slot_images_[static_cast<size_t>(slot)].payload) {
+        const SlotImages & images = slot_images_[static_cast<size_t>(slot)];
+        images.rows.overwrite(stage.embeddings.data(), stage.kv_pos, stage.chunk, b_.w_.n_embd);
+        stage.positions.assign(static_cast<size_t>(4) * stage.chunk, 0);
+        images.rows.prompt->positions.fill(stage.positions.data(), stage.kv_pos, stage.chunk);
+    }
     stage.ready = true;
     return stage;
 }
@@ -856,7 +905,7 @@ SeqEngine::StepResult Qwen35SeqEngine::step_chain_spec(
         seq_lens_[static_cast<size_t>(lane.slot)] = lane.position + 1;
         for (int axis = 0; axis < 3; ++axis) {
             positions[static_cast<size_t>(axis) * total_rows + lane_index] =
-                lane.position;
+                lane.position + rope_delta(lane.slot);
         }
         for (int head = 0; head < n_head_kv; ++head) {
             write_rows[static_cast<size_t>(head) * total_rows + lane_index] =
@@ -884,7 +933,8 @@ SeqEngine::StepResult Qwen35SeqEngine::step_chain_spec(
                 ? -1
                 : node - 1;
             query_slots[static_cast<size_t>(row)] = proposal.slot;
-            const int position = slots_.slot(proposal.slot).cur_pos + node;
+            const int position = slots_.slot(proposal.slot).cur_pos + node +
+                rope_delta(proposal.slot);
             for (int axis = 0; axis < 3; ++axis) {
                 positions[static_cast<size_t>(axis) * total_rows + row] =
                     position;
@@ -1427,14 +1477,23 @@ SeqEngine::StepResult Qwen35SeqEngine::step(const StepPlan & plan) {
     pos_buf_.assign((size_t)4 * n_total, 0);
     token_offset = 0;
     for (const PrefillStage & prefill : prefills) {
-        fill_qwen35_mrope_positions(
-            pos_buf_.data(), n_total, token_offset,
-            prefill.kv_pos, prefill.chunk);
+        if (prefill.positions.empty()) {
+            fill_qwen35_mrope_positions(
+                pos_buf_.data(), n_total, token_offset,
+                prefill.kv_pos, prefill.chunk);
+        } else {
+            for (int axis = 0; axis < 4; ++axis) {
+                std::copy_n(prefill.positions.data() + (size_t)axis * prefill.chunk, prefill.chunk,
+                            pos_buf_.data() + (size_t)axis * n_total + token_offset);
+            }
+        }
         token_offset += prefill.chunk;
     }
     if (with_decode) {
         for (int row = 0; row < live_count; ++row) {
-            const int pos = live_positions_[(size_t)row];
+            // Rotary positions run ahead of KV positions after an image.
+            const int pos = live_positions_[(size_t)row] +
+                rope_delta(live_slot_ids_[(size_t)row]);
             const int packed_row = n_prefill + row;
             pos_buf_[(size_t)0 * n_total + packed_row] = pos;
             pos_buf_[(size_t)1 * n_total + packed_row] = pos;
@@ -1650,6 +1709,7 @@ bool Qwen35SeqEngine::evict_kv(int slot, int32_t pending_token,
 
 void Qwen35SeqEngine::retire(int slot) {
     offload_.discard(slot);
+    clear_slot_images(slot);
     if (!slots_.is_active(slot)) return;
     if (slot >= 0 && slot < static_cast<int>(slot_draft_kv_.size()) &&
         slot_draft_kv_[static_cast<size_t>(slot)]) {

server/src/qwen35/concurrency/qwen35_seq_engine.h

@@ -26,6 +26,7 @@
 #include "common/dflash_draft_kv.h"
 #include "common/dflash_feature_ring.h"
 #include "qwen35_slot_manager.h"
+#include "../qwen35_image_request.h"
 
 #include <algorithm>
 #include <cstdint>
@@ -114,6 +115,11 @@ class Qwen35SeqEngine final : public SeqEngine {
         return slots_.kv_restore_feasible(slot);
     }
     void retire(int slot) override;
+    bool supports_images() const override;
+    AdmitResult admit_images(uint64_t request_id,
+                             const std::vector<int32_t> & prompt,
+                             const SamplerCfg & sampler,
+                             const ImagePromptHandle & images) override;
 
     bool token_is_eos(int32_t token) const override;
 
@@ -126,8 +132,30 @@ class Qwen35SeqEngine final : public SeqEngine {
         bool commit = false;
         std::vector<int64_t> rows;
         std::vector<float> embeddings;
+        // Axis-major [4 x chunk] rotary positions when the slot holds images;
+        // empty means the plain kv_pos + i positions.
+        std::vector<int32_t> positions;
     };
 
+    // Per-slot image state: the payload (kept alive for re-prefill after
+    // eviction), its encoded rows, and how far rotary positions run ahead of
+    // KV positions after the images.
+    struct SlotImages {
+        ImagePromptHandle payload;
+        Qwen35ImageRows rows;
+        int rope_delta = 0;
+    };
+    std::vector<SlotImages> slot_images_;
+    int rope_delta(int slot) const {
+        return slot >= 0 && slot < static_cast<int>(slot_images_.size())
+            ? slot_images_[static_cast<size_t>(slot)].rope_delta : 0;
+    }
+    void clear_slot_images(int slot) {
+        if (slot >= 0 && slot < static_cast<int>(slot_images_.size())) {
+            slot_images_[static_cast<size_t>(slot)] = SlotImages{};
+        }
+    }
+
     struct PreparedChainDraft {
         std::vector<int32_t> tokens;
     };

server/src/server/http_server.cpp

@@ -1248,7 +1248,8 @@ HttpServer::HttpServer(luce::engine::LuceEngine & engine,
                    config.disk_cache_cold_max_tokens}, backend_)
 {
     config_.image_input_enabled = backend_.supports_images() &&
-        config_.pflash_upstream_base.empty() && !backend_.seq_engine();
+        config_.pflash_upstream_base.empty() &&
+        (!backend_.seq_engine() || backend_.seq_engine()->supports_images());
     if (backend_.supports_images() && !config_.image_input_enabled) {
         std::fprintf(stderr,
             "[server] WARNING: a vision projector is loaded but image input is off: it is "

server/src/server/scheduler.cpp

@@ -451,7 +451,9 @@ void HttpServer::scheduler_loop(SeqEngine & engine) {
         PrefixCaptureTxn prepared_capture;
         PrefixCache::InlineReservation capture_reservation;
         int restore_policy_slot = -1;
-        const bool prefix_supported =
+        // Tokens alone do not identify an image: image requests never touch
+        // the prefix cache.
+        const bool prefix_supported = !req.images &&
             engine.supports_prefix_store() && !prefix_cache_.disabled();
         if (prefix_supported) {
             const auto hit = prefix_cache_.lookup_candidate(
@@ -502,7 +504,10 @@ void HttpServer::scheduler_loop(SeqEngine & engine) {
         // Admission only claims the slot and queues the prompt. Prefill
         // advances one chunk per engine step alongside live decode.
         const PrefixStorePlan requested_plan = prefix_plan;
-        auto ar = prefix_supported
+        auto ar = req.images
+            ? engine.admit_images(
+                  next_request_id, req.prompt_tokens, req.sampler, req.images)
+            : prefix_supported
             ? engine.admit_with_prefix(
                   next_request_id, req.prompt_tokens, req.sampler,
                   requested_plan)