server: refactor server_stream (#25541 )

* server: refactoring, remove spipe from server_http_res * wip * remove non-thread-safe rd.stop() call * move server_res_spipe * nits * improve server_stream_create_spipe * server-stream: update dev docs for the improved API --------- Co-authored-by: Pascal <admin@serveurperso.com>
ggml : add GGML_OP_LIGHTNING_INDEXER that implements DeepSeek V3.2/V4 lightning indexer (#24231 )
2026-07-12 15:35:53 +02:00 · 2026-07-11 12:41:47 +02:00 · 2026-07-11 11:39:07 +02:00 · 2026-07-11 10:28:29 +02:00
20 changed files with 431 additions and 211 deletions
@@ -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
@@ -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);
@@ -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;
            }
@@ -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]);
+            }
+        }
+    }
+}
@@ -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);
@@ -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;
@@ -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) {
@@ -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() {
@@ -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;
@@ -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;
@@ -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 {
@@ -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;
@@ -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));
@@ -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;
 }

@@ -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.

@@ -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;
 }

@@ -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();
    }
 }

@@ -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;
 };
@@ -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;
    };
 }
@@ -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);
+};