gguf : reject empty metadata keys (#24917)

If a Cuda device has no or limited available memory, the actual call
to cudaMemGetInfo() itself can cause a fatal crash due to a cuda out
of memory error (there is not enough memory to actually query memory)

This causes an issue because we query memory for all devices at
startup even if the user isn't trying to use the device for inference.

Fix this by making the error non-fatal and assigning zero total/free
memory to the device. This will have the downstream effect of the fit
algorithm not trying to put any layers on it, which is desired outcome
vs hard crashing.

this also prevents crashes in cuda enabled builds when user explicitly
passes '-dev none'

ggml/include/ggml-rpc.h

@@ -8,10 +8,10 @@ extern "C" {
 
 #define RPC_PROTO_MAJOR_VERSION    4
 #define RPC_PROTO_MINOR_VERSION    0
-#define RPC_PROTO_PATCH_VERSION    1
+#define RPC_PROTO_PATCH_VERSION    2
 
 #ifdef  __cplusplus
-static_assert(GGML_OP_COUNT == 97, "GGML_OP_COUNT has changed - update RPC_PROTO_PATCH_VERSION");
+static_assert(GGML_OP_COUNT == 98, "GGML_OP_COUNT has changed - update RPC_PROTO_PATCH_VERSION");
 #endif
 
 #define GGML_RPC_MAX_SERVERS       16

ggml/include/ggml.h

@@ -570,6 +570,7 @@ extern "C" {
         GGML_OP_RWKV_WKV7,
         GGML_OP_SOLVE_TRI,
         GGML_OP_GATED_DELTA_NET,
+        GGML_OP_LIGHTNING_INDEXER,
 
         GGML_OP_UNARY,
 
@@ -2575,6 +2576,24 @@ extern "C" {
             struct ggml_tensor  * state,
             int64_t               K);
 
+    // DSA lightning indexer
+    //
+    // q:       [n_embd_idx, n_head_idx, n_batch, ne3 ]
+    // k:       [n_embd_idx, 1,          n_kv,    ne3 ]
+    // weights: [n_head_idx, n_batch,    1,       ne3 ] !! prescaled !!
+    // mask:    [n_kv,       n_batch,    1,       ne33] !! f16 !!
+    // res:     [n_kv,       n_batch,    1,       ne3 ]
+    //
+    // broadcast:
+    //   ne3 % ne33 == 0
+    //
+    GGML_API struct ggml_tensor * ggml_lightning_indexer(
+        struct ggml_context * ctx,
+        struct ggml_tensor  * q,
+        struct ggml_tensor  * k,
+        struct ggml_tensor  * weights,
+        struct ggml_tensor  * mask);
+
     // custom operators
 
     typedef void (*ggml_custom1_op_t)(struct ggml_tensor * dst , const struct ggml_tensor * a, int ith, int nth, void * userdata);

ggml/src/ggml-cpu/ggml-cpu.c

@@ -2060,6 +2060,10 @@ static void ggml_compute_forward(struct ggml_compute_params * params, struct ggm
             {
                 ggml_compute_forward_gated_delta_net(params, tensor);
             } break;
+        case GGML_OP_LIGHTNING_INDEXER:
+            {
+                ggml_compute_forward_lightning_indexer(params, tensor);
+            } break;
         case GGML_OP_MAP_CUSTOM1:
             {
                 ggml_compute_forward_map_custom1(params, tensor);
@@ -2380,6 +2384,7 @@ static int ggml_get_n_tasks(struct ggml_tensor * node, int n_threads) {
         case GGML_OP_FLASH_ATTN_BACK:
         case GGML_OP_SSM_CONV:
         case GGML_OP_SSM_SCAN:
+        case GGML_OP_LIGHTNING_INDEXER:
             {
                 n_tasks = n_threads;
             } break;
@@ -2965,6 +2970,12 @@ struct ggml_cplan ggml_graph_plan(
                     {
                         GGML_ABORT("fatal error");
                     }
+                case GGML_OP_LIGHTNING_INDEXER:
+                    {
+                        // temp buffer for dequantizing lightning indexer keys
+                        const int64_t ne10 = node->src[1]->ne[0];
+                        cur += sizeof(float)*ne10*n_tasks;
+                    } break;
                 default:
                     break;
             }

ggml/src/ggml-cpu/ops.cpp

@@ -11568,3 +11568,87 @@ void ggml_compute_forward_fwht(const ggml_compute_params * params, ggml_tensor *
             }
     }
 }
+
+// ggml_compute_forward_lightning_indexer
+
+void ggml_compute_forward_lightning_indexer(
+        const ggml_compute_params * params,
+        ggml_tensor * dst) {
+
+    const ggml_tensor * q = dst->src[0];
+    const ggml_tensor * k = dst->src[1];
+    const ggml_tensor * w = dst->src[2]; // weights
+    const ggml_tensor * m = dst->src[3]; // mask
+
+    GGML_ASSERT(dst->type  == GGML_TYPE_F32);
+    GGML_ASSERT(   q->type == GGML_TYPE_F32);
+    GGML_ASSERT(   w->type == GGML_TYPE_F32);
+    GGML_ASSERT(   m->type == GGML_TYPE_F16);
+
+    GGML_TENSOR_LOCALS(int64_t, neq,  q, ne)
+    GGML_TENSOR_LOCALS(size_t,  nbq,  q, nb)
+    GGML_TENSOR_LOCALS(int64_t, nek,  k, ne)
+    GGML_TENSOR_LOCALS(size_t,  nbk,  k, nb)
+    GGML_TENSOR_LOCALS(int64_t, new,  w, ne)
+    GGML_TENSOR_LOCALS(size_t,  nbw,  w, nb)
+    GGML_TENSOR_LOCALS(int64_t, nem,  m, ne)
+    GGML_TENSOR_LOCALS(size_t,  nbm,  m, nb)
+    GGML_TENSOR_LOCALS(int64_t, ne, dst, ne)
+    GGML_TENSOR_LOCALS(size_t,  nb, dst, nb)
+
+    GGML_ASSERT( nb0 == ggml_type_size(dst->type));
+    GGML_ASSERT(nbq0 == ggml_type_size(  q->type));
+    GGML_ASSERT(nbk0 == ggml_type_size(  k->type));
+    GGML_ASSERT(nbw0 == ggml_type_size(  w->type));
+    GGML_ASSERT(nbm0 == ggml_type_size(  m->type));
+
+    const int n_embd    = q->ne[0];
+    const int n_head    = q->ne[1];
+    const int n_tokens  = q->ne[2];
+    const int n_stream  = q->ne[3];
+    const int n_kv      = k->ne[2];
+
+    ggml_to_float_t const k_to_float = ggml_get_type_traits(k->type)->to_float;
+    GGML_ASSERT((k->type == GGML_TYPE_F32 || k_to_float) && "lightning indexer: unsupported K-type");
+
+    const int nr  = n_kv;
+    const int ith = params->ith;
+    const int nth = params->nth;
+
+    // (temporary) buffer for K converted to float
+    float * k_row_f32 = (float *) params->wdata + ith*(1*n_embd + CACHE_LINE_SIZE_F32);
+
+    // rows per thread
+    const int dr = (nr + nth - 1)/nth;
+
+    // row range for this thread
+    const int ir0 = dr*ith;
+    const int ir1 = MIN(ir0 + dr, nr);
+
+    for (int s = 0; s < n_stream; ++s) {
+        for (int t = 0; t < n_tokens; ++t) {
+            const float       *   w_row =       (float *) ((char *)   w->data + t*nbw1 +        s*nbw3);
+            const ggml_fp16_t *   m_row = (ggml_fp16_t *) ((char *)   m->data + t*nbm1 + (s%nem3)*nbm3);
+            float             * dst_row =       (float *) ((char *) dst->data + t*nb1  +        s*nb3 );
+            for (int ik = ir0; ik < ir1; ++ik) {
+                char * k_row = (char *) k->data + ik*nbk2 + s*nbk3;
+                if (k_to_float) {
+                    k_to_float(k_row, k_row_f32, n_embd);
+                } else {
+                    k_row_f32 = (float *) k_row;
+                }
+                float score = 0.0f;
+                for (int h = 0; h < n_head; ++h) {
+                    // dot product of q and k for head h
+                    float qk = 0.0f;
+                    const float * q_row = (float *) ((char *) q->data + h*nbq1 + t*nbq2 + s*nbq3);
+                    ggml_vec_dot_f32(n_embd, &qk, 0, q_row, 0, k_row_f32, 0, 1);
+                    // ReLU and weights (prescaled)
+                    score += MAX(qk, 0.0f) * w_row[h];
+                }
+                // apply mask
+                dst_row[ik] = score + GGML_CPU_FP16_TO_FP32(m_row[ik]);
+            }
+        }
+    }
+}

ggml/src/ggml-cpu/ops.h

@@ -105,6 +105,7 @@ void ggml_compute_forward_rwkv_wkv7(const struct ggml_compute_params * params, s
 void ggml_compute_forward_solve_tri(const struct ggml_compute_params * params, struct ggml_tensor * dst);
 void ggml_compute_forward_gla(const struct ggml_compute_params * params, struct ggml_tensor * dst);
 void ggml_compute_forward_gated_delta_net(const struct ggml_compute_params * params, struct ggml_tensor * dst);
+void ggml_compute_forward_lightning_indexer(const struct ggml_compute_params * params, struct ggml_tensor * dst);
 void ggml_compute_forward_map_custom1(const struct ggml_compute_params * params, struct ggml_tensor * dst);
 void ggml_compute_forward_map_custom2(const struct ggml_compute_params * params, struct ggml_tensor * dst);
 void ggml_compute_forward_map_custom3(const struct ggml_compute_params * params, struct ggml_tensor * dst);

ggml/src/ggml-cuda/ggml-cuda.cu

@@ -4493,7 +4493,14 @@ static bool ggml_backend_cuda_get_available_uma_memory(long * available_memory_k
 static void ggml_backend_cuda_device_get_memory(ggml_backend_dev_t dev, size_t * free, size_t * total) {
     ggml_backend_cuda_device_context * ctx = (ggml_backend_cuda_device_context *)dev->context;
     ggml_cuda_set_device(ctx->device);
-    CUDA_CHECK(cudaMemGetInfo(free, total));
+    cudaError_t err = cudaMemGetInfo(free, total);
+    if (err != cudaSuccess) {
+        (void)cudaGetLastError();
+        GGML_LOG_WARN("%s: cudaMemGetInfo failed (%s), returning 0/0\n", __func__, cudaGetErrorString(err));
+        *free = 0;
+        *total = 0;
+        return;
+    }
 
 // ref: https://github.com/ggml-org/llama.cpp/pull/17368
 #if defined(__linux__)

ggml/src/ggml-vulkan/ggml-vulkan.cpp

@@ -6501,6 +6501,14 @@ static vk_device ggml_vk_get_device(size_t idx) {
                 device->mul_mat_id_m[i] = true;
                 device->mul_mat_id_s[i] = false;
                 break;
+            case VK_VENDOR_ID_QUALCOMM:
+                device->mul_mat_l[i] = false;
+                device->mul_mat_m[i] = true;
+                device->mul_mat_s[i] = true;
+                device->mul_mat_id_l[i] = false;
+                device->mul_mat_id_m[i] = true;
+                device->mul_mat_id_s[i] = true;
+                break;
 #endif
             default:
                 device->mul_mat_l[i] = true;

ggml/src/ggml.c

@@ -1079,6 +1079,7 @@ static const char * GGML_OP_NAME[GGML_OP_COUNT] = {
     "RWKV_WKV7",
     "SOLVE_TRI",
     "GATED_DELTA_NET",
+    "LIGHTNING_INDEXER",
 
     "UNARY",
 
@@ -1096,7 +1097,7 @@ static const char * GGML_OP_NAME[GGML_OP_COUNT] = {
     "GLU",
 };
 
-static_assert(GGML_OP_COUNT == 97, "GGML_OP_COUNT != 97");
+static_assert(GGML_OP_COUNT == 98, "GGML_OP_COUNT != 98");
 
 static const char * GGML_OP_SYMBOL[GGML_OP_COUNT] = {
     "none",
@@ -1190,6 +1191,7 @@ static const char * GGML_OP_SYMBOL[GGML_OP_COUNT] = {
     "rwkv_wkv7(r, w, k, v, a, b, s)",
     "A X = B, A triangular, solve X",
     "gated_delta_net(q, k, v, g, beta, s)",
+    "lightning_indexer(q, k, weights, mask)",
 
     "unary(x)",
 
@@ -1207,7 +1209,7 @@ static const char * GGML_OP_SYMBOL[GGML_OP_COUNT] = {
     "glu(x)",
 };
 
-static_assert(GGML_OP_COUNT == 97, "GGML_OP_COUNT != 97");
+static_assert(GGML_OP_COUNT == 98, "GGML_OP_COUNT != 98");
 
 static_assert(GGML_OP_POOL_COUNT == 2, "GGML_OP_POOL_COUNT != 2");
 
@@ -6287,6 +6289,42 @@ struct ggml_tensor * ggml_gated_delta_net(
     return result;
 }
 
+// ggml_lightning_indexer
+
+struct ggml_tensor * ggml_lightning_indexer(
+        struct ggml_context * ctx,
+        struct ggml_tensor  * q,
+        struct ggml_tensor  * k,
+        struct ggml_tensor  * weights,
+        struct ggml_tensor  * mask) {
+
+    GGML_ASSERT(       q->type == GGML_TYPE_F32);
+    GGML_ASSERT( weights->type == GGML_TYPE_F32);
+    GGML_ASSERT(    mask->type == GGML_TYPE_F16);
+    GGML_ASSERT(      q->ne[0] == k->ne[0]);
+    GGML_ASSERT(   mask->ne[0] == k->ne[2]);
+    GGML_ASSERT(      q->ne[1] == weights->ne[0]);
+    GGML_ASSERT(      k->ne[1] == 1);
+    GGML_ASSERT(   mask->ne[1] == q->ne[2]);
+    GGML_ASSERT(      q->ne[2] == weights->ne[1]);
+    GGML_ASSERT(weights->ne[2] == 1);
+    GGML_ASSERT(   mask->ne[2] == 1);
+    GGML_ASSERT(      q->ne[3] == k->ne[3]);
+    GGML_ASSERT(      k->ne[3] == weights->ne[3]);
+    GGML_ASSERT(weights->ne[3] % mask->ne[3] == 0);
+
+    int64_t ne[4] = { k->ne[2], q->ne[2], 1, q->ne[3] };
+    struct ggml_tensor * result = ggml_new_tensor(ctx, GGML_TYPE_F32, 4, ne);
+
+    result->op   = GGML_OP_LIGHTNING_INDEXER;
+    result->src[0] = q;
+    result->src[1] = k;
+    result->src[2] = weights;
+    result->src[3] = mask;
+
+    return result;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 struct ggml_hash_set ggml_hash_set_new(size_t size) {

ggml/src/gguf.cpp

@@ -557,6 +557,10 @@ static struct gguf_context * gguf_init_from_reader(const struct gguf_reader & gr
                 GGML_LOG_ERROR("%s: encountered bad_alloc error while reading key %" PRIi64 "\n", __func__, i);
                 ok = false;
             }
+            if (ok && key.empty()) {
+                GGML_LOG_ERROR("%s: key %" PRIi64 " is empty\n", __func__, i);
+                ok = false;
+            }
             for (size_t j = 0; ok && j < ctx->kv.size(); ++j) {
                 if (key == ctx->kv[j].key) {
                     GGML_LOG_ERROR("%s: duplicate key '%s' for tensors %zu and %" PRIi64 " \n", __func__, key.c_str(), j, i);

src/llama-context.cpp

@@ -55,6 +55,12 @@ static const llm_fused_op_probe llm_fused_op_gdn_ch_probe = {
     /*.n_tokens_per_seq =*/ 16,
 };
 
+static const llm_fused_op_probe llm_fused_op_lid_probe = {
+    /*.op               =*/ LLM_FUSED_OP_LIGHTNING_INDEXER,
+    /*.name             =*/ "Lightning Indexer",
+    /*.n_tokens_per_seq =*/ 1,
+};
+
 llama_context::llama_context(
         const llama_model & model,
               llama_context_params params) :
@@ -226,6 +232,9 @@ llama_context::llama_context(
     cparams.fused_gdn_ch = true;
     cparams.auto_fgdn    = true;
 
+    cparams.fused_lid    = true;
+    cparams.auto_flid    = true;
+
     // with causal attention, the batch size is limited by the context size
     cparams.n_batch = cparams.causal_attn ? std::min(cparams.n_ctx, params.n_batch) : params.n_batch;
 
@@ -522,6 +531,12 @@ void llama_context::resolve_fused_ops(const llama_memory_context_i * mctx, uint3
         resolve(llm_fused_op_gdn_ch_probe, cparams.fused_gdn_ch);
         cparams.auto_fgdn = false;
     }
+
+    if (cparams.auto_flid) {
+        LLAMA_LOG_INFO("%s: resolving fused Lightning Indexer support:\n", func);
+        resolve(llm_fused_op_lid_probe, cparams.fused_lid);
+        cparams.auto_flid = false;
+    }
 }
 
 void llama_context::sched_reserve() {

src/llama-cparams.h

@@ -41,6 +41,8 @@ struct llama_cparams {
     bool fused_gdn_ar;       // use fused gated delta net (autoregressive)
     bool fused_gdn_ch;       // use fused gated delta net (chunked)
     bool auto_fgdn;
+    bool fused_lid;          // use fused lightning indexer
+    bool auto_flid;
     bool no_perf;
     bool warmup;             // TODO: remove [TAG_LLAMA_GRAPH_NO_WARMUP]
     bool op_offload;

src/llama-graph.cpp

@@ -842,7 +842,7 @@ static void dsv4_build_comp_inputs(
         GGML_ASSERT(n_stream > 0);
         GGML_ASSERT(n_tokens%n_stream == 0);
 
-        inp.kq_mask = ggml_new_tensor_4d(ctx, cparams.flash_attn && strcmp(name, "lid") != 0 ? GGML_TYPE_F16 : GGML_TYPE_F32, plan.n_kv, n_tokens/n_stream, 1, n_stream);
+        inp.kq_mask = ggml_new_tensor_4d(ctx, (strcmp(name, "lid") != 0 && cparams.flash_attn) || (strcmp(name, "lid") == 0 && cparams.fused_lid) ? GGML_TYPE_F16 : GGML_TYPE_F32, plan.n_kv, n_tokens/n_stream, 1, n_stream);
         ggml_set_input(inp.kq_mask);
         ggml_set_name(inp.kq_mask, (std::string("dsv4_") + name + "_kq_mask").c_str());
     }
@@ -3025,9 +3025,9 @@ llm_graph_input_attn_k_dsa * llm_graph_context::build_attn_inp_k_dsa() const {
     {
         inp->self_k_idxs_lid = mctx_cur->get_lid()->build_input_k_idxs(ctx0, ubatch);
 
-        // ensure F32 mask
+        // ensure that mask type matches fused lightning indexer use (requires f16 mask)
         auto cparams_copy = cparams;
-        cparams_copy.flash_attn = false;
+        cparams_copy.flash_attn = cparams.fused_lid;
 
         inp->self_kq_mask_lid = build_attn_inp_kq_mask(ctx0, mctx_cur->get_lid(), ubatch, cparams_copy);
         inp->self_kq_mask_lid_cnv = inp->self_kq_mask_lid;

src/llama-graph.h

@@ -42,6 +42,7 @@ enum llm_fused_op {
     LLM_FUSED_OP_FLASH_ATTN,
     LLM_FUSED_OP_GDN_AR,
     LLM_FUSED_OP_GDN_CH,
+    LLM_FUSED_OP_LIGHTNING_INDEXER,
 };
 
 enum llm_ffn_op_type : int {

src/llama-kv-cache-dsv4.cpp

@@ -29,6 +29,15 @@ static uint32_t dsv4_comp_size(uint32_t kv_size, uint32_t ratio) {
     return std::max<uint32_t>(1, (kv_size + ratio - 1)/ratio);
 }
 
+static void dsv4_clear_tensor_stream(ggml_tensor * tensor, uint32_t stream) {
+    GGML_ASSERT(ggml_is_contiguous(tensor));
+    GGML_ASSERT(tensor->ne[3] == 1);
+    GGML_ASSERT(stream < (uint32_t) tensor->ne[2]);
+
+    const size_t stream_size = tensor->nb[2];
+    ggml_backend_tensor_memset(tensor, 0, stream*stream_size, stream_size);
+}
+
 static int64_t dsv4_stream_offset(uint32_t n_stream, llama_seq_id seq_id, uint32_t size) {
     if (n_stream <= 1) {
         return 0;
@@ -781,11 +790,20 @@ llama_dsv4_comp_state::llama_dsv4_comp_state(
             __func__, name, ratio, state_size, n_embd_state, n_stream, layers.size(), total_size()/1024.0/1024.0);
 }
 
-void llama_dsv4_comp_state::clear(bool data) {
+void llama_dsv4_comp_state::clear(llama_seq_id seq_id, bool data) {
     if (!data) {
         return;
     }
 
+    if (seq_id >= 0) {
+        GGML_ASSERT((uint32_t) seq_id < n_stream);
+        for (const auto & layer : layers) {
+            dsv4_clear_tensor_stream(layer.kv,    (uint32_t) seq_id);
+            dsv4_clear_tensor_stream(layer.score, (uint32_t) seq_id);
+        }
+        return;
+    }
+
     for (auto & [_, buf] : ctxs_bufs) {
         ggml_backend_buffer_clear(buf.get(), 0);
     }
@@ -1034,7 +1052,7 @@ llama_kv_cache_dsv4::llama_kv_cache_dsv4(
     // graph does not necessarily overwrite; uninitialized buffer contents would
     // otherwise leak in (instance-specific garbage) and corrupt recall. Zero all
     // compressed buffers up front so reads of un-written rows are deterministic.
-    clear_compressed(true);
+    clear_compressed(-1, true);
 }
 
 llama_memory_context_ptr llama_kv_cache_dsv4::init_batch(
@@ -1147,7 +1165,7 @@ bool llama_kv_cache_dsv4::get_can_shift() const {
 
 void llama_kv_cache_dsv4::clear(bool data) {
     kv_raw->clear(data);
-    clear_compressed(true); // DSV4 compressed buffers must never expose stale/uninit rows
+    clear_compressed(-1, true); // DSV4 compressed buffers must never expose stale/uninit rows
 }
 
 bool llama_kv_cache_dsv4::seq_rm(llama_seq_id seq_id, llama_pos p0, llama_pos p1) {
@@ -1169,7 +1187,7 @@ bool llama_kv_cache_dsv4::seq_rm(llama_seq_id seq_id, llama_pos p0, llama_pos p1
     const bool res = kv_raw->seq_rm(seq_id, p0, p1);
 
     if (res) {
-        clear_compressed(true);
+        clear_compressed(seq_id, true);
     }
 
     return res;
@@ -1177,22 +1195,29 @@ bool llama_kv_cache_dsv4::seq_rm(llama_seq_id seq_id, llama_pos p0, llama_pos p1
 
 void llama_kv_cache_dsv4::seq_cp(llama_seq_id seq_id_src, llama_seq_id seq_id_dst, llama_pos p0, llama_pos p1) {
     kv_raw->seq_cp(seq_id_src, seq_id_dst, p0, p1);
-    clear_compressed(true);
 }
 
 void llama_kv_cache_dsv4::seq_keep(llama_seq_id seq_id) {
+    GGML_ASSERT(seq_id >= 0 && (uint32_t) seq_id < n_seq_max);
+
     kv_raw->seq_keep(seq_id);
-    clear_compressed(true);
+
+    for (llama_seq_id id = 0; id < (llama_seq_id) n_seq_max; ++id) {
+        if (id == seq_id) {
+            continue;
+        }
+
+        kv_raw->seq_rm(id, -1, -1);
+        clear_compressed(id, true);
+    }
 }
 
 void llama_kv_cache_dsv4::seq_add(llama_seq_id seq_id, llama_pos p0, llama_pos p1, llama_pos shift) {
     kv_raw->seq_add(seq_id, p0, p1, shift);
-    clear_compressed(true);
 }
 
 void llama_kv_cache_dsv4::seq_div(llama_seq_id seq_id, llama_pos p0, llama_pos p1, int d) {
     kv_raw->seq_div(seq_id, p0, p1, d);
-    clear_compressed(true);
 }
 
 llama_pos llama_kv_cache_dsv4::seq_pos_min(llama_seq_id seq_id) const {
@@ -1328,13 +1353,32 @@ llama_dsv4_comp_state * llama_kv_cache_dsv4::get_lid_state() const {
     return lid_state.get();
 }
 
-void llama_kv_cache_dsv4::clear_compressed(bool data) {
-    kv_csa->clear(data);
-    kv_hca->clear(data);
-    kv_lid->clear(data);
-    csa_state->clear(data);
-    hca_state->clear(data);
-    lid_state->clear(data);
+void llama_kv_cache_dsv4::clear_compressed(llama_seq_id seq_id, bool data) {
+    if (seq_id < 0) {
+        kv_csa->clear(data);
+        kv_hca->clear(data);
+        kv_lid->clear(data);
+    } else {
+        GGML_ASSERT((uint32_t) seq_id < n_seq_max);
+
+        const auto clear_seq = [seq_id, data](llama_kv_cache * kv) {
+            kv->seq_rm(seq_id, -1, -1);
+
+            if (data) {
+                for (uint32_t il : kv->get_layer_ids()) {
+                    dsv4_clear_tensor_stream(kv->get_k_storage(il), (uint32_t) seq_id);
+                }
+            }
+        };
+
+        clear_seq(kv_csa.get());
+        clear_seq(kv_hca.get());
+        clear_seq(kv_lid.get());
+    }
+
+    csa_state->clear(seq_id, data);
+    hca_state->clear(seq_id, data);
+    lid_state->clear(seq_id, data);
 }
 
 //

src/llama-kv-cache-dsv4.h

@@ -21,7 +21,7 @@ public:
             const char    * name,
         const llama_memory_i::layer_filter_cb & filter);
 
-    void clear(bool data);
+    void clear(llama_seq_id seq_id, bool data);
 
     uint32_t get_ratio()    const;
     uint32_t get_state_size() const;
@@ -67,6 +67,8 @@ private:
 // DSV4 uses a normal raw/SWA token cache plus compressed K-only block caches.
 // The compressed caches are storage only; DSV4-specific visibility and block
 // planning are handled by llama_kv_cache_dsv4_context / llm_graph_input_dsv4.
+// FIXME: currently the cache only supports non-unified mode even if unified flag is passed
+// FIXME: we currently conflate token_pos and buffer contents. See https://github.com/ggml-org/llama.cpp/pull/25521#discussion_r3558173819
 
 class llama_kv_cache_dsv4 : public llama_memory_i {
 public:
@@ -146,7 +148,7 @@ private:
     std::unique_ptr<llama_dsv4_comp_state> hca_state;
     std::unique_ptr<llama_dsv4_comp_state> lid_state;
 
-    void clear_compressed(bool data);
+    void clear_compressed(llama_seq_id seq_id, bool data);
 };
 
 // DSV4 raw attention only uses the SWA half of kv_raw. The base half is kept

src/models/deepseek32.cpp

@@ -301,43 +301,50 @@ llama_model_deepseek32::graph::graph(const llama_model & model, const llm_graph_
                 indexer_q = ggml_view_4d(ctx0, indexer_q, indexer_q->ne[0], indexer_q->ne[1], indexer_q->ne[2]/n_stream, n_stream, indexer_q->nb[1], indexer_q->nb[2], indexer_q->nb[3]/n_stream, 0);
                 indexer_weights = ggml_view_4d(ctx0, indexer_weights, indexer_weights->ne[0], indexer_weights->ne[1]/n_stream, indexer_weights->ne[2], n_stream, indexer_weights->nb[1], indexer_weights->nb[2]/n_stream, indexer_weights->nb[3]/n_stream, 0);
 
-                // calculate indexer kq
-                indexer_q = ggml_permute(ctx0, indexer_q, 0, 2, 1, 3);
-                cb(indexer_q, "indexer_q", il);
-                indexer_k = ggml_permute(ctx0, indexer_k, 0, 2, 1, 3);
-                cb(indexer_k, "indexer_k", il);
-
-                ggml_tensor * indexer_kq = ggml_mul_mat(ctx0, indexer_k, indexer_q);
-                cb(indexer_kq, "indexer_kq", il);
-
-                // ReLU requires contiguous tensors
-                indexer_kq = ggml_cont(ctx0, ggml_permute(ctx0, indexer_kq, 2, 1, 0, 3));
-                cb(indexer_kq, "indexer_kq", il);
-
-                // apply ReLU
-                ggml_tensor * indexer_score = ggml_relu(ctx0, indexer_kq);
-                cb(indexer_score, "indexer_score", il);
-
                 // pre-scale weights to avoid scaling operations on huge indexer_score tensor
                 indexer_weights = ggml_scale(ctx0, indexer_weights, 1.0f / sqrtf(float(n_embd_indexer_head * n_indexer_head)));
                 cb(indexer_weights, "indexer_weights", il);
 
-                // multiply scores by indexer weights
-                indexer_score = ggml_mul(ctx0, indexer_score, indexer_weights);
-                cb(indexer_score, "indexer_score", il);
+                ggml_tensor * indexer_score = nullptr;
+                if (cparams.fused_lid) {
+                    indexer_score = ggml_lightning_indexer(ctx0, indexer_q, indexer_k, indexer_weights, inp_attn_dsa->get_kq_mask_lid());
+                    cb(indexer_score, "indexer_score", il);
+                    res->add_fused_node({LLM_FUSED_OP_LIGHTNING_INDEXER, indexer_score, il});
+                } else {
+                    // calculate indexer kq
+                    indexer_q = ggml_permute(ctx0, indexer_q, 0, 2, 1, 3);
+                    cb(indexer_q, "indexer_q", il);
+                    indexer_k = ggml_permute(ctx0, indexer_k, 0, 2, 1, 3);
+                    cb(indexer_k, "indexer_k", il);
 
-                // sum by q n_indexer_head dimension
-                indexer_score = ggml_sum_rows(ctx0, indexer_score);
-                cb(indexer_score, "indexer_score", il);
+                    ggml_tensor * indexer_kq = ggml_mul_mat(ctx0, indexer_k, indexer_q);
+                    cb(indexer_kq, "indexer_kq", il);
 
-                // permute result to match KQ mask
-                indexer_score = ggml_cont(ctx0, ggml_permute(ctx0, indexer_score, 2, 1, 0, 3));
-                cb(indexer_score, "indexer_score", il);
+                    // ReLU requires contiguous tensors
+                    indexer_kq = ggml_cont(ctx0, ggml_permute(ctx0, indexer_kq, 2, 1, 0, 3));
+                    cb(indexer_kq, "indexer_kq", il);
 
-                // mask indexer scores
-                ggml_tensor * indexer_kq_mask = inp_attn_dsa->get_kq_mask_lid();
-                indexer_score = ggml_add(ctx0, indexer_score, indexer_kq_mask);
-                cb(indexer_score, "indexer_score", il);
+                    // apply ReLU
+                    indexer_score = ggml_relu(ctx0, indexer_kq);
+                    cb(indexer_score, "indexer_score", il);
+
+                    // multiply scores by indexer weights
+                    indexer_score = ggml_mul(ctx0, indexer_score, indexer_weights);
+                    cb(indexer_score, "indexer_score", il);
+
+                    // sum by q n_indexer_head dimension
+                    indexer_score = ggml_sum_rows(ctx0, indexer_score);
+                    cb(indexer_score, "indexer_score", il);
+
+                    // permute result to match KQ mask
+                    indexer_score = ggml_cont(ctx0, ggml_permute(ctx0, indexer_score, 2, 1, 0, 3));
+                    cb(indexer_score, "indexer_score", il);
+
+                    // mask indexer scores
+                    ggml_tensor * indexer_kq_mask = inp_attn_dsa->get_kq_mask_lid();
+                    indexer_score = ggml_add(ctx0, indexer_score, indexer_kq_mask);
+                    cb(indexer_score, "indexer_score", il);
+                }
 
                 // get indices of top k indexer scores
                 uint32_t n_top_k = indexer_score->ne[0] < n_indexer_top_k ? indexer_score->ne[0] : n_indexer_top_k;

src/models/deepseek4.cpp

@@ -556,25 +556,32 @@ ggml_tensor * llama_model_deepseek4::graph::build_lid_top_k(
             indexer_weights->ne[0], indexer_weights->ne[1]/n_stream, indexer_weights->ne[2], n_stream,
             indexer_weights->nb[1], indexer_weights->nb[2]/n_stream, indexer_weights->nb[3]/n_stream, 0);
 
-    indexer_q = ggml_permute(ctx0, indexer_q, 0, 2, 1, 3);
-    cb(indexer_q, "lid_q", il);
-    indexer_k = ggml_permute(ctx0, indexer_k, 0, 2, 1, 3);
-    cb(indexer_k, "lid_k", il);
+    ggml_tensor * indexer_score = nullptr;
+    if (cparams.fused_lid) {
+        indexer_score = ggml_lightning_indexer(ctx0, indexer_q, indexer_k, indexer_weights, inp_lid.kq_mask);
+        cb(indexer_score, "lid_score_masked", il);
+        res->add_fused_node({LLM_FUSED_OP_LIGHTNING_INDEXER, indexer_score, il});
+    } else {
+        indexer_q = ggml_permute(ctx0, indexer_q, 0, 2, 1, 3);
+        cb(indexer_q, "lid_q", il);
+        indexer_k = ggml_permute(ctx0, indexer_k, 0, 2, 1, 3);
+        cb(indexer_k, "lid_k", il);
 
-    ggml_tensor * indexer_kq = ggml_mul_mat(ctx0, indexer_k, indexer_q);
-    cb(indexer_kq, "lid_kq", il);
+        ggml_tensor * indexer_kq = ggml_mul_mat(ctx0, indexer_k, indexer_q);
+        cb(indexer_kq, "lid_kq", il);
 
-    indexer_kq = ggml_cont(ctx0, ggml_permute(ctx0, indexer_kq, 2, 1, 0, 3));
-    cb(indexer_kq, "lid_kq", il);
+        indexer_kq = ggml_cont(ctx0, ggml_permute(ctx0, indexer_kq, 2, 1, 0, 3));
+        cb(indexer_kq, "lid_kq", il);
 
-    ggml_tensor * indexer_score = ggml_relu(ctx0, indexer_kq);
-    indexer_score = ggml_mul(ctx0, indexer_score, indexer_weights);
-    indexer_score = ggml_sum_rows(ctx0, indexer_score);
-    indexer_score = ggml_cont(ctx0, ggml_permute(ctx0, indexer_score, 2, 1, 0, 3));
-    cb(indexer_score, "lid_score", il);
+        indexer_score = ggml_relu(ctx0, indexer_kq);
+        indexer_score = ggml_mul(ctx0, indexer_score, indexer_weights);
+        indexer_score = ggml_sum_rows(ctx0, indexer_score);
+        indexer_score = ggml_cont(ctx0, ggml_permute(ctx0, indexer_score, 2, 1, 0, 3));
+        cb(indexer_score, "lid_score", il);
 
-    indexer_score = ggml_add(ctx0, indexer_score, inp_lid.kq_mask);
-    cb(indexer_score, "lid_score_masked", il);
+        indexer_score = ggml_add(ctx0, indexer_score, inp_lid.kq_mask);
+        cb(indexer_score, "lid_score_masked", il);
+    }
 
     const uint32_t n_top_k = indexer_score->ne[0] < hparams.indexer_top_k ? indexer_score->ne[0] : hparams.indexer_top_k;
     ggml_tensor * top_k = ggml_cont(ctx0, ggml_top_k(ctx0, indexer_score, n_top_k));

tests/test-backend-ops.cpp

@@ -7097,6 +7097,67 @@ struct test_diag : public test_case {
     }
 };
 
+// GGML_OP_LIGHTNING_INDEXER
+struct test_lightning_indexer : public test_case {
+    const int64_t hsk; // indexer K head size
+    const int64_t nh; // num indexer heads
+    const int64_t kv; // kv size
+    const int64_t nb; // batch size
+    const int64_t ns; // num streams
+    const int64_t nm; // ne[3] of mask
+
+    const ggml_type type_K;
+
+    std::string vars() override {
+        return VARS_TO_STR7(hsk, nh, kv, nb, ns, nm, type_K);
+    }
+
+    double max_nmse_err() override {
+        return 1e-6;
+    }
+
+    uint64_t op_flops(ggml_tensor * t) override {
+        GGML_UNUSED(t);
+        return ((2 * hsk + 2) * nh + 1) * kv * nb * ns;
+    }
+
+    test_lightning_indexer(int64_t hsk = 128, int64_t nh = 64, int64_t kv = 256, int64_t nb = 128, int64_t ns = 1, int64_t nm = 1, ggml_type type_K = GGML_TYPE_F16)
+        : hsk(hsk), nh(nh), kv(kv), nb(nb), ns(ns), nm(nm), type_K(type_K) {}
+
+    ggml_tensor * build_graph(ggml_context * ctx) override {
+        ggml_tensor * q = ggml_new_tensor_4d(ctx, GGML_TYPE_F32, hsk, nh, nb, ns);
+        ggml_set_param(q);
+        ggml_set_name(q, "q");
+
+        ggml_tensor * k = ggml_new_tensor_4d(ctx, type_K, hsk, 1, kv, ns);
+        ggml_set_param(k);
+        ggml_set_name(k, "k");
+
+        ggml_tensor * w = ggml_new_tensor_4d(ctx, GGML_TYPE_F32, nh, nb, 1, ns);
+        ggml_set_param(w);
+        ggml_set_name(w, "w");
+
+        ggml_tensor * m = ggml_new_tensor_4d(ctx, GGML_TYPE_F16, kv, nb, 1, nm);
+        ggml_set_param(m);
+        ggml_set_name(m, "m");
+
+        ggml_tensor * out = ggml_lightning_indexer(ctx, q, k, w, m);
+        ggml_set_name(out, "out");
+
+        return out;
+    }
+
+    void initialize_tensors(ggml_context * ctx) override {
+        for (ggml_tensor * t = ggml_get_first_tensor(ctx); t != NULL; t = ggml_get_next_tensor(ctx, t)) {
+            if (strcmp(t->name, "m") == 0) {
+                init_tensor_kq_mask(t);
+            } else {
+                init_tensor_uniform(t);
+            }
+        }
+    }
+};
+
 // Deserializable generic test case
 struct input_tensor {
     ggml_type type;
@@ -9393,6 +9454,19 @@ static std::vector<std::unique_ptr<test_case>> make_test_cases_eval() {
     test_cases.emplace_back(new test_falcon(2));
 #endif
 
+    // lightning_indexer
+    for (int kv : { 256 }) {
+        for (int bs : { 1, 512 }) {
+            for (int nh : { 32, 64 }) {
+                for (auto [ns, nm] : { std::pair{1, 1}, std::pair{4, 4}, std::pair{4, 1} }) {
+                    for (ggml_type type_K : {GGML_TYPE_F32, GGML_TYPE_F16, GGML_TYPE_BF16, GGML_TYPE_Q8_0, GGML_TYPE_Q5_1, GGML_TYPE_Q5_0, GGML_TYPE_Q4_1, GGML_TYPE_Q4_0, GGML_TYPE_IQ4_NL}) {
+                        test_cases.emplace_back(new test_lightning_indexer(128, nh, kv, bs, ns, nm, type_K));
+                    }
+                }
+            }
+        }
+    }
+
     return test_cases;
 }
 #ifdef _MSC_VER
@@ -9722,6 +9796,19 @@ static std::vector<std::unique_ptr<test_case>> make_test_cases_perf() {
     test_cases.emplace_back(new test_gated_delta_net(GGML_TYPE_F32, 4, 128, 1024, 1)); // 4h PP-1024
     test_cases.emplace_back(new test_gated_delta_net(GGML_TYPE_F32, 32, 128, 64, 1, 1, false, true)); // KDA PP-64
 
+    // lightning_indexer
+    for (int kv : { 256, 4096, 65536 }) {
+        for (int bs : { 1, 512, 2048 }) {
+            for (int nh : { 32, 64 }) {
+                for (int ns : { 1, 4 }) {
+                    for (ggml_type type_K : {GGML_TYPE_F32, GGML_TYPE_F16, GGML_TYPE_BF16, GGML_TYPE_Q8_0, GGML_TYPE_Q5_1, GGML_TYPE_Q5_0, GGML_TYPE_Q4_1, GGML_TYPE_Q4_0, GGML_TYPE_IQ4_NL}) {
+                        test_cases.emplace_back(new test_lightning_indexer(128, nh, kv, bs, ns, ns, type_K));
+                    }
+                }
+            }
+        }
+    }
+
     return test_cases;
 }
 

tests/test-gguf.cpp

@@ -26,6 +26,7 @@ enum handcrafted_file_type {
     HANDCRAFTED_HEADER_EMPTY               = 800,
 
     HANDCRAFTED_KV_BAD_KEY_SIZE            =  10 + offset_has_kv,
+    HANDCRAFTED_KV_EMPTY_KEY               =  15 + offset_has_kv,
     HANDCRAFTED_KV_BAD_TYPE                =  20 + offset_has_kv,
     // HANDCRAFTED_KV_BAD_VALUE_SIZE          =  30 + offset_has_kv, // removed because it can result in allocations > 1 TB (default sanitizer limit)
     HANDCRAFTED_KV_DUPLICATE_KEY           =  40 + offset_has_kv,
@@ -64,6 +65,7 @@ static std::string handcrafted_file_type_name(const enum handcrafted_file_type h
         case HANDCRAFTED_HEADER_EMPTY:               return "HEADER_EMPTY";
 
         case HANDCRAFTED_KV_BAD_KEY_SIZE:            return "KV_BAD_KEY_SIZE";
+        case HANDCRAFTED_KV_EMPTY_KEY:               return "KV_EMPTY_KEY";
         case HANDCRAFTED_KV_BAD_TYPE:                return "KV_BAD_TYPE";
         case HANDCRAFTED_KV_DUPLICATE_KEY:           return "KV_DUPLICATE_KEY";
         case HANDCRAFTED_KV_BAD_ALIGN:               return "KV_BAD_ALIGN";
@@ -284,7 +286,9 @@ static FILE * get_handcrafted_file(const unsigned int seed, const enum handcraft
         const enum gguf_type type     = gguf_type(hft == HANDCRAFTED_KV_BAD_TYPE ? GGUF_TYPE_COUNT : kv_types[i].first);
         const enum gguf_type type_arr = gguf_type(hft == HANDCRAFTED_KV_BAD_TYPE ? GGUF_TYPE_COUNT : kv_types[i].second);
 
-        const std::string key = "my_key_" + std::to_string((hft == HANDCRAFTED_KV_DUPLICATE_KEY ? i/2 : i));
+        const std::string key = hft == HANDCRAFTED_KV_EMPTY_KEY
+            ? ""
+            : "my_key_" + std::to_string((hft == HANDCRAFTED_KV_DUPLICATE_KEY ? i/2 : i));
 
         if (hft == HANDCRAFTED_KV_BAD_KEY_SIZE) {
             const uint64_t n = -1;
@@ -732,6 +736,7 @@ static std::pair<int, int> test_handcrafted_file(const unsigned int seed) {
         HANDCRAFTED_HEADER_EMPTY,
 
         HANDCRAFTED_KV_BAD_KEY_SIZE,
+        HANDCRAFTED_KV_EMPTY_KEY,
         HANDCRAFTED_KV_BAD_TYPE,
         HANDCRAFTED_KV_DUPLICATE_KEY,
         HANDCRAFTED_KV_BAD_ALIGN,

tests/test-save-load-state.cpp

@@ -78,7 +78,84 @@ static llama_tokens test_baseline(struct llama_model * model, const struct commo
 }
 
 
-// Test 2: state load
+// Test 2: sequence removal isolation
+// - decode the same prefix into two sequences
+// - remove sequence 0
+// - verify that sequence 1 remains unchanged
+static bool test_seq_rm_isolated(
+        struct llama_model         * model,
+        const struct common_params & params,
+        const llama_tokens         & tokens) {
+    auto params_ctx = common_context_params_to_llama(params);
+    params_ctx.n_ctx      = 256;
+    params_ctx.n_seq_max  = 2;
+    params_ctx.kv_unified = true;
+
+    auto ctx = llama_context_ptr{llama_init_from_model(model, params_ctx)};
+    if (!ctx) {
+        LOG_ERR("%s: failed to create context\n", __func__);
+        return false;
+    }
+
+    LOG("\n=== Test 2: sequence removal isolation ===\n");
+
+    const size_t n_tokens = tokens.size() < 128 ? tokens.size() : 128;
+    for (llama_seq_id seq_id = 0; seq_id < 2; ++seq_id) {
+        llama_batch_ptr batch(n_tokens, 0, 1);
+        for (size_t i = 0; i < n_tokens; ++i) {
+            common_batch_add(batch.get(), tokens[i], i, { seq_id }, false);
+        }
+
+        if (llama_decode(ctx.get(), batch.get())) {
+            LOG_ERR("%s: failed to decode prompt for sequence %d\n", __func__, seq_id);
+            return false;
+        }
+    }
+
+    const auto get_seq_state = [&](llama_seq_id seq_id, std::vector<uint8_t> & state) {
+        const size_t state_size = llama_state_seq_get_size(ctx.get(), seq_id);
+        if (state_size == 0) {
+            LOG_ERR("%s: sequence state is empty\n", __func__);
+            return false;
+        }
+
+        state.resize(state_size);
+        const size_t ncopy = llama_state_seq_get_data(ctx.get(), state.data(), state.size(), seq_id);
+        if (ncopy != state.size()) {
+            LOG_ERR("%s: sequence state length %zu does not match expected length %zu\n",
+                    __func__, ncopy, state.size());
+            return false;
+        }
+
+        return true;
+    };
+
+    std::vector<uint8_t> state_before;
+    if (!get_seq_state(1, state_before)) {
+        return false;
+    }
+
+    if (!llama_memory_seq_rm(llama_get_memory(ctx.get()), 0, -1, -1)) {
+        LOG_ERR("%s: failed to remove sequence 0\n", __func__);
+        return false;
+    }
+
+    std::vector<uint8_t> state_after;
+    if (!get_seq_state(1, state_after)) {
+        return false;
+    }
+
+    if (state_before != state_after) {
+        LOG_ERR("%s: removing sequence 0 changed sequence 1\n", __func__);
+        return false;
+    }
+
+    LOG("PASS\n");
+    return true;
+}
+
+
+// Test 3: state load
 // - create a new context
 // - load state from file
 // - replay the last prompt token
@@ -90,7 +167,7 @@ static bool test_state_load(struct llama_model * model, const struct common_para
     auto smpl = llama_sampler_ptr{llama_sampler_chain_init(sparams)};
     llama_sampler_chain_add(smpl.get(), llama_sampler_init_dist(params.sampling.seed));
 
-    LOG("\n=== Test 2: state load ===\n");
+    LOG("\n=== Test 3: state load ===\n");
 
     // Load state from file
     llama_tokens unused_sts(tokens.size());
@@ -126,7 +203,7 @@ static bool test_state_load(struct llama_model * model, const struct common_para
 }
 
 
-// Test 3: seq copy (host)
+// Test 4: seq copy (host)
 // - create a multi-seq context
 // - load state from file
 // - replay the last prompt token
@@ -141,7 +218,7 @@ static bool test_seq_cp_host(struct llama_model * model, const struct common_par
     auto smpl = llama_sampler_ptr{llama_sampler_chain_init(sparams)};
     llama_sampler_chain_add(smpl.get(), llama_sampler_init_dist(params.sampling.seed));
 
-    LOG("\n=== Test 3: seq copy (host) ===\n");
+    LOG("\n=== Test 4: seq copy (host) ===\n");
 
     // Load state from file
     llama_tokens unused_sts(tokens.size());
@@ -198,7 +275,7 @@ static bool test_seq_cp_host(struct llama_model * model, const struct common_par
 }
 
 
-// Test 4: seq copy (device)
+// Test 5: seq copy (device)
 // - create a multi-seq context
 // - load state from file
 // - replay the last prompt token
@@ -213,7 +290,7 @@ static bool test_seq_cp_device(struct llama_model * model, const struct common_p
     auto smpl = llama_sampler_ptr{llama_sampler_chain_init(sparams)};
     llama_sampler_chain_add(smpl.get(), llama_sampler_init_dist(params.sampling.seed));
 
-    LOG("\n=== Test 4: seq copy (device) ===\n");
+    LOG("\n=== Test 5: seq copy (device) ===\n");
 
     // Load state from file
     llama_tokens unused_sts(tokens.size());
@@ -337,17 +414,22 @@ int main(int argc, char ** argv) {
         return 1;
     }
 
-    // Test 2: state load
+    // Test 2: sequence removal isolation
+    if (!test_seq_rm_isolated(model, params, tokens)) {
+        return 1;
+    }
+
+    // Test 3: state load
     if (!test_state_load(model, params, tokens, result_baseline)) {
         return 1;
     }
 
-    // Test 3: seq copy (host)
+    // Test 4: seq copy (host)
     if (!test_seq_cp_host(model, params, tokens, result_baseline)) {
         return 1;
     }
 
-    // Test 4: seq copy (device)
+    // Test 5: seq copy (device)
     if (!test_seq_cp_device(model, params, tokens, result_baseline)) {
         return 1;
     }

tools/server/README-dev.md

@@ -126,15 +126,15 @@ It is opt in via the `X-Conversation-Id` header on `POST /v1/chat/completions`.
 
 The feature lives entirely in `server-stream.{h,cpp}` and rests on three types:
 
-- `stream_session`: a bounded ring buffer (4 MiB cap, oldest bytes drop first) plus a condvar. `append` pushes raw SSE bytes, `read_from` drains from any offset and blocks for live bytes or finalize, `finalize` wakes readers, `cancel` stops the producer. One conv maps to at most one live session.
+- `stream_session`: a bounded ring buffer (4 MiB cap, oldest bytes drop first) plus a condvar. `append` pushes raw SSE bytes, `read_from` drains from any offset and blocks for live bytes or finalize, `finalize` wakes readers, `cancel` sets the flag the producer polls. One conv maps to at most one live session.
 - `stream_session_manager`: a file-static singleton (`g_stream_sessions`) inside `server-stream.cpp`, owns all sessions keyed by conv id, enforces the one conv one session invariant via `create_or_replace`, and runs a GC thread that drops completed sessions past their TTL. Exposed to main only through `server_stream_session_manager_start/stop`.
 - `stream_pipe_producer` / `stream_pipe_consumer`: the write and read ends. The producer owns the session lifetime and finalizes it on destruction; the consumer is read only and never finalizes, so a reader detaching cannot kill a running generation.
 
-The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes only the route handler factories, `server_stream_session_attach_pipe`, `server_stream_aware_should_stop`, `server_stream_conv_id_from_headers` and the GC lifecycle; the session, manager and consumer types stay in the `.cpp`.
+The implementation is hidden in `server-stream.cpp` (pimpl). The header exposes only the route handler factories, the `server_res_spipe` response base, `server_stream_conv_id_from_headers` and the GC lifecycle; the session, manager, consumer and the `server_stream_create_spipe` factory stay in the `.cpp`.
 
-Producer side: `server_res_generator` attaches a producer pipe when the header is present. The HTTP content provider mirrors every chunk into the ring before writing it to the socket. While a pipe is attached, `server_stream_aware_should_stop` ignores peer disconnect, so a dropped socket does not stop generation: only an explicit `DELETE` does. When the peer leaves early, `on_complete` calls `close()`, which drains the rest of the generation into the ring on the http worker.
+Producer side: `server_res_generator` extends `server_res_spipe`, which keeps all spipe logic out of the generic `server_http_res`. `set_req` attaches a producer when the header is present, and the wrapped `next` tees each chunk into the ring before the socket, so a chunk lost to a dead wire is already buffered. While attached, `should_stop` ignores peer disconnect: only a `DELETE` stops generation. On an early peer drop, `on_complete` drains the tail into the ring on the http worker.
 
-Lifetime safety: the producer pipe holds a shared `alive` flag also captured by the session cancel hook. `~server_res_generator` calls `cleanup()` to clear that hook while the reader is still alive, so a `cancel` arriving during teardown can never call `stop()` on a freed response. This ordering is the most fragile part of the feature: finalizing or destroying the producer before `cleanup()` runs reintroduces a use after free.
+Lifetime safety: the session holds no back reference to the response, so `spipe` is a plain `unique_ptr` touched only by the http worker. `cancel` raises an atomic the producer polls; the producer finalizes the session from its destructor, which also runs `~server_response_reader::stop()` to cancel the generation at the queue level. A `DELETE` stops work by raising the flag and letting the worker unwind.
 
 Consumer side: `GET /v1/stream/<conv_id>?from=N` opens a `text/event-stream` that replays buffered bytes from offset `N` and blocks for live bytes, so the browser reattaches like a fresh EventSource. An offset below the dropped prefix returns 400.
 
@@ -235,6 +235,29 @@ That requires `JSON.stringify` when formatted to message content:
 }
 ```
 
+Set `stream: true` in the request body to stream a tool's output as it runs, instead of waiting for it to finish. Only certain tools accept this (for ex. `exec_shell_command`);
+returns 404 if tool doesn't support it.
+
+Response is SSE stream, one `data: <json>` line per chunk:
+
+```json
+{"chunk": "hello\n"}
+```
+
+followed by a final event once the tool returns:
+
+```json
+{"done": true}
+```
+
+or, if `invoke()` threw:
+
+```json
+{"done": true, "error": "..."}
+```
+
+There is no `[DONE]` sentinel (unlike `/chat/completions`), the stream ends after the `done`
+
 ### Router mode: how child <--> router communicates
 
 Upon spawning a new child process using `subprocess`, both child and router listen to the stdout/stderr (combined)

tools/server/server-context.cpp

@@ -3979,11 +3979,9 @@ server_context_meta server_context::get_meta() const {
     };
 }
 
-
-
 // generator-like API for HTTP response generation
 // may have bypass_sleep = true if the task does not use ctx_server
-struct server_res_generator : server_http_res {
+struct server_res_generator : server_res_spipe {
     server_response_reader rd;
     server_res_generator(server_queue & queue_tasks, server_response & queue_results, int sleep_idle_seconds, bool bypass_sleep = false)
             : rd(queue_tasks, queue_results, HTTP_POLLING_SECONDS) {
@@ -3993,15 +3991,6 @@ struct server_res_generator : server_http_res {
             queue_tasks.wait_until_no_sleep();
         }
     }
-    ~server_res_generator() override {
-        // cleanup() must run while rd is still alive (rd is destroyed after this body returns)
-        if (spipe) {
-            spipe->cleanup();
-        }
-    }
-    void stop() override {
-        rd.stop();
-    }
     void ok(const json & response_data) {
         status = 200;
         data = safe_json_to_str(response_data);
@@ -4039,6 +4028,8 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
     auto & rd = res->rd;
     auto & params = this->params;
 
+    res->set_req(&req); // will also set spipe if needed
+
     int32_t sse_ping_interval = params.sse_ping_interval;
 
     try {
@@ -4181,7 +4172,7 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
         }
         res->status = 200;
         res->content_type = "text/event-stream";
-        res->next = [res_this = res.get(), res_type, sse_ping_interval, &req](std::string & output) -> bool {
+        res->set_next([res_this = res.get(), res_type, sse_ping_interval](std::string & output) -> bool {
             static auto format_error = [](task_response_type res_type, const json & res_json) {
                 if (res_type == TASK_RESPONSE_TYPE_ANTHROPIC) {
                     return format_anthropic_sse({
@@ -4193,7 +4184,9 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
                 }
             };
 
-            auto effective_should_stop = server_stream_aware_should_stop(res_this, req.should_stop);
+            auto effective_should_stop = [&res_this]() {
+                return res_this->should_stop();
+            };
 
             try {
                 if (effective_should_stop()) {
@@ -4284,13 +4277,9 @@ std::unique_ptr<server_res_generator> server_routes::handle_completions_impl(
                 // terminate on exception
                 return false;
             }
-        };
+        });
     }
 
-    // attach a producer pipe to the response when X-Conversation-Id is present.
-    // the pipe mirrors SSE chunks into the ring buffer and wires up the cancel hook.
-    server_stream_session_attach_pipe(*res, req.headers);
-
     return res;
 }
 

tools/server/server-http.cpp

@@ -1,7 +1,6 @@
 #include "common.h"
 #include "http.h"
 #include "server-http.h"
-#include "server-stream.h"
 #include "server-common.h"
 #include "ui.h"
 
@@ -530,33 +529,20 @@ static void process_handler_response(server_http_req_ptr && request, server_http
             std::string chunk;
             const bool has_next = response->next(chunk);
             if (!chunk.empty()) {
-                // mirror into the ring buffer first, the session must reflect every SSE chunk
-                // whether or not the wire write below succeeds
-                if (response->spipe) {
-                    response->spipe->write(chunk.data(), chunk.size());
-                }
                 if (!sink.write(chunk.data(), chunk.size())) {
-                    // peer is gone, stop the wire path here
                     return false;
                 }
                 SRV_DBG("http: streamed chunk: %s\n", chunk.c_str());
             }
             if (!has_next) {
-                // producer reached its natural end on the wire, a later close() skips the drain
-                if (response->spipe) {
-                    response->spipe->done();
-                }
                 sink.done();
                 SRV_DBG("%s", "http: stream ended\n");
             }
             return has_next;
         };
         const auto on_complete = [request = q_ptr, response = r_ptr](bool) mutable {
-            // on a dropped peer, close() drains the rest of the generation into the ring buffer
-            if (response->spipe) {
-                response->spipe->close();
-            }
-            response.reset(); // spipe destructor finalizes the session if attached
+            response->on_complete();
+            response.reset();
             request.reset();
         };
         res.set_chunked_content_provider(content_type, chunked_content_provider, on_complete);
@@ -564,6 +550,7 @@ static void process_handler_response(server_http_req_ptr && request, server_http
         res.status = response->status;
         set_headers(res, response->headers);
         res.set_content(response->data, response->content_type);
+        response->on_complete();
     }
 }
 

tools/server/server-http.h

@@ -11,7 +11,6 @@
 #include <unordered_map>
 
 struct common_params;
-struct stream_pipe_producer; // defined in server-stream.h
 
 // generator-like API for HTTP response generation
 // this object response with one of the 2 modes:
@@ -25,19 +24,13 @@ struct server_http_res {
     std::string data;
     std::map<std::string, std::string> headers;
 
-    // if set, the stream survives a client disconnect: the producer pipe keeps draining into the
-    // ring buffer and finalizes the session on destruction, so no explicit on_stream_end is needed.
-    // shared_ptr (not unique_ptr) so the forward-declared type is safe to delete here.
-    std::shared_ptr<stream_pipe_producer> spipe;
-
     std::function<bool(std::string &)> next = nullptr;
     bool is_stream() const {
         return next != nullptr;
     }
 
-    // called when the session is cancelled (e.g. DELETE /v1/stream/<conv_id>).
-    // server_res_generator overrides this to stop its reader; the default is a no-op.
-    virtual void stop() {}
+    // fired before req and res are destroyed
+    virtual void on_complete() {}
 
     virtual ~server_http_res() = default;
 };

tools/server/server-stream.cpp

@@ -96,8 +96,6 @@ struct stream_session {
     size_t  dropped_prefix() const; // bytes evicted from the front due to cap
     int64_t completed_at() const;   // 0 while alive, unix seconds after finalize
 
-    void set_stop_producer(std::function<void()> fn);
-
     void cancel();
 
 private:
@@ -109,7 +107,6 @@ private:
     bool                    done;
     std::atomic<bool>       cancelled; // polled lock-free by the should_stop closure, no mu
     int64_t                 completed_ts;
-    std::function<void()>   stop_producer;
 };
 stream_session::stream_session(std::string conversation_id_, size_t max_bytes_)
     : conversation_id(std::move(conversation_id_))
@@ -217,26 +214,10 @@ int64_t stream_session::completed_at() const {
     return completed_ts;
 }
 
-void stream_session::set_stop_producer(std::function<void()> fn) {
-    std::lock_guard<std::mutex> lock(mu);
-    stop_producer = std::move(fn);
-}
-
 void stream_session::cancel() {
-    // flip cancelled first so the producer-side server_stream_aware_should_stop can break out of the
-    // recv() wait even if remove_waiting_task_ids does not notify the condvar (the cancel task
-    // posted by rd.stop() will eventually notify, but we do not want to depend on that timing)
+    // the should_stop closure on both the producer and any HTTP reader polls is_cancelled()
+    // so flipping this is the only signal needed to unwind both sides
     cancelled.store(true, std::memory_order_release);
-    // copy the hook under the lock then invoke outside, the producer side may grab queue locks
-    // and we do not want to hold our mu across that path
-    std::function<void()> fn;
-    {
-        std::lock_guard<std::mutex> lock(mu);
-        fn = stop_producer;
-    }
-    if (fn) {
-        fn();
-    }
 }
 
 bool stream_session::is_cancelled() const {
@@ -325,8 +306,10 @@ void stream_session_manager::evict_and_cancel(const std::string & conversation_i
         s = it->second;
         sessions.erase(it);
     }
-    // signal the producer side first so the inference is cancelled at the queue level,
-    // then finalize, which wakes any pending HTTP reader and lets the drain exit naturally
+    // cancel first so the producer's on_complete() drain loop and any pending HTTP reader
+    // observe is_cancelled() and stop pulling further output, then finalize to wake readers
+    // blocked in read_from(). note: this does not interrupt the underlying generation itself,
+    // which keeps running to its own natural stop condition (EOS/max_tokens)
     s->cancel();
     s->finalize();
 }
@@ -431,65 +414,15 @@ stream_pipe_producer::stream_pipe_producer(stream_session_ptr session)
 }
 
 stream_pipe_producer::~stream_pipe_producer() {
-    cleanup();
     session_->finalize();
 }
 
-void stream_pipe_producer::cleanup() {
-    if (!alive_) {
-        return;
-    }
-    alive_->store(false, std::memory_order_release);
-    session_->set_stop_producer(nullptr);
-    alive_.reset();
-}
-
 bool stream_pipe_producer::write(const char * data, size_t len) {
     return session_->append(data, len);
 }
 
-void stream_pipe_producer::done() {
-    done_ = true;
-}
-
-void stream_pipe_producer::close() {
-    // httplib bails its content provider the moment is_peer_alive() goes false, so pump the rest
-    // of the generation into the ring buffer here. a DELETE flips is_cancelled and cuts it short
-    if (done_ || session_->is_cancelled()) {
-        SRV_TRC("stream_pipe close: skip drain (done=%d cancelled=%d) conv=%s\n",
-                done_ ? 1 : 0, session_->is_cancelled() ? 1 : 0, session_->conversation_id.c_str());
-        return;
-    }
-    SRV_TRC("stream_pipe close: draining conv=%s\n", session_->conversation_id.c_str());
-    size_t drained = 0;
-    std::string chunk;
-    while (true) {
-        chunk.clear();
-        bool has_next = res_->next(chunk);
-        if (!chunk.empty()) {
-            write(chunk.data(), chunk.size());
-            drained += chunk.size();
-        }
-        if (!has_next) {
-            break;
-        }
-    }
-    SRV_TRC("stream_pipe close: drain ended conv=%s bytes=%zu\n", session_->conversation_id.c_str(), drained);
-}
-
-std::shared_ptr<stream_pipe_producer> stream_pipe_producer::create(stream_session_ptr session,
-                                                                   server_http_res & res) {
-    auto alive = std::make_shared<std::atomic<bool>>(true);
-    auto * res_ptr = &res;
-    session->set_stop_producer([alive, res_ptr]() {
-        if (alive->load(std::memory_order_acquire)) {
-            res_ptr->stop();
-        }
-    });
-    auto pipe = std::shared_ptr<stream_pipe_producer>(new stream_pipe_producer(std::move(session)));
-    pipe->alive_ = std::move(alive);
-    pipe->res_   = res_ptr;
-    return pipe;
+stream_pipe_producer * stream_pipe_producer::create(stream_session_ptr session) {
+    return new stream_pipe_producer(std::move(session));
 }
 
 // stream_pipe_consumer
@@ -661,21 +594,68 @@ std::string server_stream_conv_id_from_headers(const std::map<std::string, std::
     return std::string();
 }
 
-void server_stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & headers) {
+static stream_pipe_producer * server_stream_create_spipe(const std::map<std::string, std::string> & headers) {
     std::string conversation_id = server_stream_conv_id_from_headers(headers);
     SRV_TRC("conv_id=%s (empty=%d)\n", conversation_id.c_str(), conversation_id.empty() ? 1 : 0);
     if (conversation_id.empty()) {
-        return;
+        return nullptr;
     }
     auto session = g_stream_sessions.create_or_replace(conversation_id);
-    res.spipe = stream_pipe_producer::create(session, res);
+    return stream_pipe_producer::create(session);
 }
 
-std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback) {
-    return [res, fallback = std::move(fallback)]() -> bool {
-        if (res->spipe) {
-            return res->spipe->is_cancelled();
+//
+// server_res_spipe
+//
+
+void server_res_spipe::set_req(const server_http_req * req) {
+    this->req = req;
+    // optionally attach spipe to the response when X-Conversation-Id is present
+    spipe.reset(server_stream_create_spipe(req->headers));
+}
+
+bool server_res_spipe::conn_alive() {
+    GGML_ASSERT(req != nullptr);
+    return !req->should_stop();
+}
+
+bool server_res_spipe::should_stop() {
+    if (spipe) {
+        // note: if DELETE /v1/stream/<conv_id> is called, is_cancelled() will be true
+        return spipe->is_cancelled();
+    } else {
+        return !conn_alive();
+    }
+}
+
+void server_res_spipe::on_complete() {
+    if (!spipe || next_finished) {
+        return;
+    }
+    std::string chunk;
+    while (!spipe->is_cancelled()) {
+        chunk.clear();
+        bool has_next = next_orig(chunk);
+        if (!chunk.empty()) {
+            spipe->write(chunk.data(), chunk.size());
         }
-        return fallback();
+        if (!has_next) {
+            break;
+        }
+    }
+}
+
+void server_res_spipe::set_next(std::function<bool(std::string &)> next_fn) {
+    next_orig = std::move(next_fn);
+    next = [this](std::string & out) {
+        bool has_next = next_orig(out);
+        if (spipe) {
+            // if spipe is set, tee-style pipe input to both HTTP and spipe
+            spipe->write(out.data(), out.size());
+        }
+        if (!has_next) {
+            next_finished = true;
+        }
+        return has_next;
     };
 }

tools/server/server-stream.h

@@ -30,36 +30,15 @@ protected:
 
 // producer end: writes chunks into the ring buffer and owns the session lifetime, finalizing it
 // on destruction.
-//
-// lifetime safety: holds a shared_ptr<atomic<bool>> alive also captured by the session's
-// stop_producer hook. cleanup() sets alive=false and clears the hook; it must run while the
-// response the hook calls stop() on is still alive. ~server_res_generator() does this explicitly.
 struct stream_pipe_producer : stream_pipe {
     ~stream_pipe_producer() override;
 
     bool write(const char * data, size_t len);
 
-    // mark the natural end on the wire so a later close() is a no-op
-    void done();
-
-    // on a peer drop, pump the response next() into the ring buffer until done. runs on the http
-    // worker from on_complete, no-op after done() or cancel
-    void close();
-
-    // disarm the stop hook and drop the alive guard, must run while the response the hook
-    // references is still alive. idempotent, the destructor calls it too
-    void cleanup();
-
-    // res.stop() is invoked when the session is cancelled, the alive guard ensures stop() is not
-    // called after cleanup() has run
-    static std::shared_ptr<stream_pipe_producer> create(stream_session_ptr session, server_http_res & res);
+    static stream_pipe_producer * create(stream_session_ptr session);
 
 private:
     explicit stream_pipe_producer(stream_session_ptr session);
-
-    bool                                done_ = false;
-    std::shared_ptr<std::atomic<bool>>  alive_;
-    server_http_res *                   res_ = nullptr;
 };
 
 void server_stream_session_manager_start();
@@ -73,10 +52,22 @@ server_http_context::handler_t server_stream_make_delete_handler();
 // extract the X-Conversation-Id header value (case-insensitive), empty when absent
 std::string server_stream_conv_id_from_headers(const std::map<std::string, std::string> & headers);
 
-// on an X-Conversation-Id header, create or replace the session and attach a producer pipe to res
-void server_stream_session_attach_pipe(server_http_res & res, const std::map<std::string, std::string> & headers);
+// implement tee-style pipe (spipe) for "stream replay" functionality
+struct server_res_spipe : server_http_res {
+private:
+    // if set, the stream survives a client disconnect:
+    // connection kept alive, output is forwarded to spipe and reuse later
+    std::unique_ptr<stream_pipe_producer> spipe;
+    // if spipe is set, use this next_orig to implement tee-style pipe
+    std::function<bool(std::string &)> next_orig;
+    const server_http_req * req = nullptr;
+    // set once next_orig reports no more data, so on_complete() doesn't re-drain a finished stream
+    bool next_finished = false;
 
-// should_stop closure that ignores peer disconnect when a pipe is attached, so only an explicit
-// DELETE stops the producer and generation keeps flowing into the ring buffer. without a pipe it
-// delegates to fallback, the legacy non-resumable flow
-std::function<bool()> server_stream_aware_should_stop(server_http_res * res, std::function<bool()> fallback);
+public:
+    void set_req(const server_http_req * req);
+    bool conn_alive();
+    bool should_stop();
+    void on_complete() override;
+    void set_next(std::function<bool(std::string &)> next_fn);
+};

tools/server/server-tools.cpp

@@ -12,6 +12,7 @@
 #include <climits>
 #include <algorithm>
 #include <unordered_set>
+#include <functional>
 
 namespace fs = std::filesystem;
 
@@ -51,7 +52,13 @@ public:
     virtual bool write_file(const std::string & path, const std::string & content) const = 0;
     // paths relative to `base`, '/'-separated; sets `err` if `base` isn't a directory
     virtual std::vector<std::string> list_files(const std::string & base, std::string & err) const = 0;
-    virtual exec_result run(const std::vector<std::string> & args, size_t max_output, int timeout_secs) const = 0;
+    // on_chunk, if set, is called with each chunk of output as it is read (before truncation cuts in);
+    // returning false terminates the process early (e.g. the client disconnected)
+    virtual exec_result run(
+            const std::vector<std::string> & args,
+            size_t max_output,
+            int timeout_secs,
+            const std::function<bool(const std::string &)> & on_chunk = nullptr) const = 0;
 };
 
 class tools_io_basic : public tools_io {
@@ -123,7 +130,11 @@ public:
         return list_files_fallback(base);
     }
 
-    exec_result run(const std::vector<std::string> & args, size_t max_output, int timeout_secs) const override {
+    exec_result run(
+            const std::vector<std::string> & args,
+            size_t max_output,
+            int timeout_secs,
+            const std::function<bool(const std::string &)> & on_chunk = nullptr) const override {
         exec_result res;
 
         subprocess_s proc;
@@ -164,8 +175,14 @@ public:
                     size_t len = strlen(buf);
                     if (output.size() + len <= max_output) {
                         output.append(buf, len);
+                        if (on_chunk && !on_chunk(std::string(buf, len))) {
+                            subprocess_terminate(&proc);
+                            break;
+                        }
                     } else {
-                        output.append(buf, max_output - output.size());
+                        size_t remaining = max_output - output.size();
+                        output.append(buf, remaining);
+                        if (on_chunk && remaining > 0) on_chunk(std::string(buf, remaining));
                         truncated = true;
                     }
                 }
@@ -287,7 +304,7 @@ struct server_tool_read_file : server_tool {
         };
     }
 
-    json invoke(json params) const override {
+    json invoke(json params, server_tool::stream *) const override {
         std::string path  = params.at("path").get<std::string>();
         int  start_line   = json_value(params, "start_line", 1);
         int  end_line     = json_value(params, "end_line",  -1); // -1 = no limit
@@ -376,7 +393,7 @@ struct server_tool_file_glob_search : server_tool {
         };
     }
 
-    json invoke(json params) const override {
+    json invoke(json params, server_tool::stream *) const override {
         std::string base    = params.at("path").get<std::string>();
         std::string include = json_value(params, "include", std::string("**"));
         std::string exclude = json_value(params, "exclude", std::string(""));
@@ -457,7 +474,7 @@ struct server_tool_grep_search : server_tool {
         };
     }
 
-    json invoke(json params) const override {
+    json invoke(json params, server_tool::stream *) const override {
         std::string path        = params.at("path").get<std::string>();
         std::string pat_str     = params.at("pattern").get<std::string>();
         std::string include     = json_value(params, "include", std::string("**"));
@@ -577,6 +594,7 @@ struct server_tool_exec_shell_command : server_tool {
         name = "exec_shell_command";
         display_name = "Execute shell command";
         permission_write = true;
+        support_stream = true;
     }
 
     json get_definition() const override {
@@ -598,7 +616,7 @@ struct server_tool_exec_shell_command : server_tool {
         };
     }
 
-    json invoke(json params) const override {
+    json invoke(json params, server_tool::stream * st) const override {
         std::string command   = params.at("command").get<std::string>();
         int    timeout        = json_value(params, "timeout",         10);
         size_t max_output     = (size_t) json_value(params, "max_output_size", (int) SERVER_TOOL_EXEC_SHELL_COMMAND_MAX_OUTPUT_SIZE);
@@ -612,7 +630,24 @@ struct server_tool_exec_shell_command : server_tool {
         std::vector<std::string> args = {"sh", "-c", command};
 #endif
 
-        auto io  = make_tools_io(params);
+        auto io = make_tools_io(params);
+
+        if (st) {
+            auto res = io->run(args, max_output, timeout, [st](const std::string & chunk) {
+                st->push(chunk);
+                return !st->alive || st->alive();
+            });
+            if (st->alive && !st->alive()) {
+                return json();
+            }
+            std::string tail = string_format("\n[exit code: %d]", res.exit_code);
+            if (res.timed_out) {
+                tail += " [exit due to timed out]";
+            }
+            st->push(tail);
+            return json();
+        }
+
         auto res = io->run(args, max_output, timeout);
 
         std::string text_output = res.output;
@@ -654,7 +689,7 @@ struct server_tool_write_file : server_tool {
         };
     }
 
-    json invoke(json params) const override {
+    json invoke(json params, server_tool::stream *) const override {
         std::string path    = params.at("path").get<std::string>();
         std::string content = params.at("content").get<std::string>();
 
@@ -710,7 +745,7 @@ struct server_tool_edit_file : server_tool {
         };
     }
 
-    json invoke(json params) const override {
+    json invoke(json params, server_tool::stream *) const override {
         std::string path = params.at("path").get<std::string>();
         const json & edits_json = params.at("edits");
 
@@ -1018,7 +1053,7 @@ struct server_tool_get_datetime : server_tool {
         };
     }
 
-    json invoke(json) const override {
+    json invoke(json, server_tool::stream *) const override {
         auto now = std::chrono::system_clock::now();
         auto time = std::chrono::system_clock::to_time_t(now);
 
@@ -1026,6 +1061,59 @@ struct server_tool_get_datetime : server_tool {
     }
 };
 
+struct server_tool_stream_result : server_task_result {
+    std::string chunk;
+    bool done = false;
+    std::string error_msg;
+
+    json to_json() override {
+        if (!done) {
+            return {{"chunk", chunk}};
+        } else {
+            json result = {{"done", true}};
+            if (!error_msg.empty()) {
+                result["error"] = error_msg;
+            }
+            return result;
+        }
+    }
+};
+
+void server_tool::stream::push(const std::string & chunk) {
+    if (chunk.empty()) return;
+    auto r = std::make_unique<server_tool_stream_result>();
+    r->id    = id;
+    r->chunk = chunk;
+    qr.send(std::move(r));
+}
+
+struct server_tools_res : server_http_res {
+    std::thread worker;
+    server_response * qr = nullptr; // set only for streaming responses
+    int id = -1;
+
+    ~server_tools_res() override {
+        if (worker.joinable()) {
+            worker.join();
+        }
+        if (qr) {
+            qr->remove_waiting_task_id(id);
+        }
+    }
+};
+
+static server_tool & find_tool(std::vector<std::unique_ptr<server_tool>> & tools, const std::string & name, bool require_stream) {
+    for (auto & t : tools) {
+        if (t->name == name) {
+            if (require_stream && !t->support_stream) {
+                throw std::invalid_argument(string_format("tool \"%s\" does not support stream = true", name.c_str()));
+            }
+            return *t;
+        }
+    }
+    throw std::invalid_argument(string_format("unknown tool \"%s\"", name.c_str()));
+}
+
 //
 // public API
 //
@@ -1090,16 +1178,63 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools) {
     };
 
     handle_post = [this](const server_http_req & req) -> server_http_res_ptr {
-        auto res = std::make_unique<server_http_res>();
+        auto res = std::make_unique<server_tools_res>();
         try {
             json body = json::parse(req.body);
             std::string tool_name = body.at("tool").get<std::string>();
             json params = body.value("params", json::object());
-            json result = invoke(tool_name, params);
-            res->data   = safe_json_to_str(result);
+            bool stream = body.value("stream", false);
+
+            server_tool & tool = find_tool(tools, tool_name, stream);
+
+            if (stream) {
+                int id = res_id.fetch_add(1);
+                queue_res.add_waiting_task_id(id);
+                res->qr = &queue_res;
+                res->id = id;
+
+                res->worker = std::thread([this, id, &req, &tool, params]() mutable {
+                    server_tool::stream st{queue_res, id, [&req]() {
+                        return !req.should_stop();
+                    }};
+
+                    auto done = std::make_unique<server_tool_stream_result>();
+                    try {
+                        tool.invoke(params, &st);
+                    } catch (const std::exception & e) {
+                        done->error_msg = e.what();
+                    } catch (...) {
+                        done->error_msg = "An unknown error occurred";
+                    }
+                    done->id    = st.id;
+                    done->done  = true;
+                    st.qr.send(std::move(done));
+                });
+
+                res->content_type = "text/event-stream";
+                res->status = 200;
+                res->next   = [this, id](std::string & output) -> bool {
+                    auto result = queue_res.recv(id);
+                    auto * r = dynamic_cast<server_tool_stream_result *>(result.get());
+                    GGML_ASSERT(r != nullptr);
+                    output = "data: " + safe_json_to_str(r->to_json()) + "\n\n";
+                    if (r->done) {
+                        queue_res.remove_waiting_task_id(id);
+                        return false;
+                    }
+                    return true;
+                };
+            } else {
+                json result = tool.invoke(params, nullptr);
+                res->status = 200;
+                res->data   = safe_json_to_str(result);
+            }
         } catch (const json::exception & e) {
             res->status = 400;
             res->data   = safe_json_to_str(format_error_response(e.what(), ERROR_TYPE_INVALID_REQUEST));
+        } catch (const std::invalid_argument & e) {
+            res->status = 404;
+            res->data   = safe_json_to_str(format_error_response(e.what(), ERROR_TYPE_INVALID_REQUEST));
         } catch (const std::exception & e) {
             SRV_ERR("got exception: %s\n", e.what());
             res->status = 500;
@@ -1108,12 +1243,3 @@ void server_tools::setup(const std::vector<std::string> & enabled_tools) {
         return res;
     };
 }
-
-json server_tools::invoke(const std::string & name, const json & params) {
-    for (auto & t : tools) {
-        if (t->name == name) {
-            return t->invoke(params);
-        }
-    }
-    return {{"error", "unknown tool: " + name}};
-}

tools/server/server-tools.h

@@ -2,15 +2,27 @@
 
 #include "server-common.h"
 #include "server-http.h"
+#include "server-queue.h"
+
+#include <atomic>
+#include <functional>
 
 struct server_tool {
     std::string name;
     std::string display_name;
     bool permission_write = false;
+    bool support_stream = false; // if true, output can be streamed
 
     virtual ~server_tool() = default;
     virtual json get_definition() const = 0;
-    virtual json invoke(json params) const = 0;
+
+    struct stream {
+        server_response & qr;
+        int id;
+        std::function<bool()> alive;
+        void push(const std::string & chunk);
+    };
+    virtual json invoke(json params, stream * st = nullptr) const = 0;
 
     json to_json() const;
 };
@@ -18,8 +30,11 @@ struct server_tool {
 struct server_tools {
     std::vector<std::unique_ptr<server_tool>> tools;
 
+    // for streaming
+    server_response queue_res;
+    std::atomic<int> res_id{0};
+
     void setup(const std::vector<std::string> & enabled_tools);
-    json invoke(const std::string & name, const json & params);
 
     server_http_context::handler_t handle_get;
     server_http_context::handler_t handle_post;

tools/server/tests/unit/test_tools_builtin.py

@@ -105,6 +105,24 @@ def test_tools_builtin_edit_file_rejects_non_unique_old_text():
             os.remove(log_path)
 
 
+def test_tools_builtin_exec_shell_command_stream():
+    global server
+    server.start()
+
+    events = list(server.make_stream_request("POST", "/tools", data={
+        "tool": "exec_shell_command",
+        "params": {"command": "echo hello"},
+        "stream": True,
+    }))
+
+    assert len(events) >= 2
+    assert events[-1]["done"] is True
+    assert not events[-1].get("error")
+    chunks = "".join(e["chunk"] for e in events[:-1])
+    assert "hello" in chunks
+    assert "[exit code: 0]" in chunks
+
+
 def test_tools_builtin_edit_file_rejects_overlapping_edits():
     global server
     server.start()