## Coverage Report

#### Created: 2026-01-25 15:05

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /home/noah/src/realizar/src/api/gpu_handlers.rs

    Line
    Count
    Source
    1
    //! GPU batch inference handlers
    2
    //!
    3
    //! Extracted from api/mod.rs (PMAT-802) to reduce module size.
    4
    //! Contains batch completions, warmup, and status handlers for GPU inference.
    5
    6
    use std::convert::Infallible;
    7
    8
    use axum::{
    9
        extract::State,
    10
        http::StatusCode,
    11
        response::sse::{Event, Sse},
    12
        Json,
    13
    };
    14
    use futures::stream::Stream;
    15
    use serde::{Deserialize, Serialize};
    16
    17
    use super::{
    18
        AppState, ErrorResponse, GenerateRequest, GenerateResponse, BatchGenerateResponse,
    19
        StreamTokenEvent, StreamDoneEvent, ModelsResponse, TokenizeRequest, TokenizeResponse,
    20
        BatchTokenizeRequest, BatchTokenizeResponse, BatchGenerateRequest,
    21
        default_max_tokens, default_top_k,
    22
    };
    23
    use crate::generate::{GenerationConfig, SamplingStrategy};
    24
    use crate::registry::ModelInfo;
    25
    26
    // ============================================================================
    27
    // PARITY-022: GPU Batch Inference API
    28
    // ============================================================================
    29
    30
    /// GPU batch completions request (PARITY-022)
    31
    #[derive(Debug, Clone, Serialize, Deserialize)]
    32
    pub struct GpuBatchRequest {
    33
        /// List of prompts to process in batch
    34
        pub prompts: Vec<String>,
    35
        /// Maximum tokens to generate per prompt
    36
        #[serde(default = "default_max_tokens")]
    37
        pub max_tokens: usize,
    38
        /// Temperature for sampling (0.0 = greedy)
    39
        #[serde(default)]
    40
        pub temperature: f32,
    41
        /// Top-k sampling (1 = greedy)
    42
        #[serde(default = "default_top_k")]
    43
        pub top_k: usize,
    44
        /// Stop tokens (optional)
    45
        #[serde(default)]
    46
        pub stop: Vec<String>,
    47
    }
    48
    49
    /// GPU batch completions response (PARITY-022)
    50
    #[derive(Debug, Clone, Serialize, Deserialize)]
    51
    pub struct GpuBatchResponse {
    52
        /// Results for each prompt
    53
        pub results: Vec<GpuBatchResult>,
    54
        /// Batch statistics
    55
        pub stats: GpuBatchStats,
    56
    }
    57
    58
    /// Single result in GPU batch response
    59
    #[derive(Debug, Clone, Serialize, Deserialize)]
    60
    pub struct GpuBatchResult {
    61
        /// Prompt index
    62
        pub index: usize,
    63
        /// Generated token IDs
    64
        pub token_ids: Vec<u32>,
    65
        /// Decoded text
    66
        pub text: String,
    67
        /// Number of tokens generated
    68
        pub num_generated: usize,
    69
    }
    70
    71
    /// GPU batch statistics
    72
    #[derive(Debug, Clone, Serialize, Deserialize)]
    73
    pub struct GpuBatchStats {
    74
        /// Batch size
    75
        pub batch_size: usize,
    76
        /// Whether GPU was used
    77
        pub gpu_used: bool,
    78
        /// Total tokens generated
    79
        pub total_tokens: usize,
    80
        /// Processing time in milliseconds
    81
        pub processing_time_ms: f64,
    82
        /// Throughput in tokens per second
    83
        pub throughput_tps: f64,
    84
    }
    85
    86
    /// GPU warmup response (PARITY-022)
    87
    #[derive(Debug, Clone, Serialize, Deserialize)]
    88
    pub struct GpuWarmupResponse {
    89
        /// Whether warmup succeeded
    90
        pub success: bool,
    91
        /// Memory used in bytes
    92
        pub memory_bytes: usize,
    93
        /// Number of layers cached
    94
        pub num_layers: usize,
    95
        /// Message
    96
        pub message: String,
    97
    }
    98
    99
    /// GPU status response (PARITY-022)
    100
    #[derive(Debug, Clone, Serialize, Deserialize)]
    101
    pub struct GpuStatusResponse {
    102
        /// Whether GPU cache is warmed up
    103
        pub cache_ready: bool,
    104
        /// Memory used by cache in bytes
    105
        pub cache_memory_bytes: usize,
    106
        /// GPU batch threshold
    107
        pub batch_threshold: usize,
    108
        /// Recommended minimum batch size
    109
        pub recommended_min_batch: usize,
    110
    }
    111
    112
    // ==================== PARITY-052: Batch Request Queuing ====================
    113
    //
    114
    // Infrastructure for continuous batch inference via HTTP API.
    115
    // Requests are queued and processed in batches for higher throughput.
    116
    //
    117
    // Architecture:
    118
    //   - BatchConfig: Configuration for batch window and size thresholds
    119
    //   - ContinuousBatchRequest: Internal request with oneshot response channel
    120
    //   - ContinuousBatchResponse: Result returned via oneshot channel
    121
    //   - AppState extensions: batch_scheduler, batch_request_tx, batch_config
    122
    // ============================================================================
    123
    124
    /// Configuration for continuous batch inference (PARITY-052)
    125
    #[derive(Debug, Clone)]
    126
    #[cfg(feature = "gpu")]
    127
    pub struct BatchConfig {
    128
        /// Maximum time to wait for batch to fill (milliseconds)
    129
        pub window_ms: u64,
    130
        /// Minimum batch size to process (below this, use single-request path)
    131
        pub min_batch: usize,
    132
        /// Optimal batch size for M4 parity (process immediately when reached)
    133
        /// PARITY-095: This also controls GPU batch threshold
    134
        pub optimal_batch: usize,
    135
        /// Maximum batch size (GPU memory constraint)
    136
        pub max_batch: usize,
    137
        /// Channel buffer size for request queue
    138
        pub queue_size: usize,
    139
        /// GPU batch threshold (use GPU path when batch >= this)
    140
        /// PARITY-095: GPU GEMM wins at batch >= 32 (from IMP-600 analysis)
    141
        pub gpu_threshold: usize,
    142
    }
    143
    144
    #[cfg(feature = "gpu")]
    145
    impl Default for BatchConfig {
    146
    5
        fn default() -> Self {
    147
    5
            Self {
    148
    5
                window_ms: 50,     // 50ms batch window (allow time for requests to accumulate)
    149
    5
                min_batch: 4,      // Minimum for any batching benefit
    150
    5
                optimal_batch: 32, // PARITY-095: Aligned with GPU threshold for M4 parity
    151
    5
                max_batch: 64,     // Allow larger batches for better GPU utilization
    152
    5
                queue_size: 1024,  // Request queue buffer
    153
    5
                gpu_threshold: 32, // GPU GEMM crossover point (from PARITY-046b)
    154
    5
            }
    155
    5
        }
    156
    }
    157
    158
    #[cfg(feature = "gpu")]
    159
    impl BatchConfig {
    160
        /// Create config optimized for low latency (smaller batches)
    161
        /// Note: GPU batch disabled (threshold > max_batch) for consistent latency
    162
    2
        pub fn low_latency() -> Self {
    163
    2
            Self {
    164
    2
                window_ms: 5,
    165
    2
                min_batch: 2,
    166
    2
                optimal_batch: 8,
    167
    2
                max_batch: 16,
    168
    2
                queue_size: 512,
    169
    2
                gpu_threshold: 32, // Effectively disabled since max_batch=16
    170
    2
            }
    171
    2
        }
    172
    173
        /// Create config optimized for high throughput (larger batches)
    174
        /// PARITY-095: GPU batch enabled for batch >= 32
    175
    2
        pub fn high_throughput() -> Self {
    176
    2
            Self {
    177
    2
                window_ms: 100, // 100ms window for maximum batching
    178
    2
                min_batch: 8,
    179
    2
                optimal_batch: 32, // Trigger processing at GPU threshold
    180
    2
                max_batch: 128,    // Large batches for maximum throughput
    181
    2
                queue_size: 2048,
    182
    2
                gpu_threshold: 32, // GPU GEMM crossover
    183
    2
            }
    184
    2
        }
    185
    186
        /// Check if batch size is sufficient for processing
    187
    12
        pub fn should_process(&self, batch_size: usize) -> bool {
    188
    12
            batch_size >= self.optimal_batch
    189
    12
        }
    190
    191
        /// Check if batch size meets minimum threshold
    192
    11
        pub fn meets_minimum(&self, batch_size: usize) -> bool {
    193
    11
            batch_size >= self.min_batch
    194
    11
        }
    195
    }
    196
    197
    /// Internal batch request with response channel (PARITY-052)
    198
    #[cfg(feature = "gpu")]
    199
    pub struct ContinuousBatchRequest {
    200
        /// Tokenized input prompt
    201
        pub prompt_tokens: Vec<u32>,
    202
        /// Maximum tokens to generate
    203
        pub max_tokens: usize,
    204
        /// Sampling temperature
    205
        pub temperature: f32,
    206
        /// Top-k sampling parameter
    207
        pub top_k: usize,
    208
        /// Channel to send response back to handler
    209
        pub response_tx: tokio::sync::oneshot::Sender<ContinuousBatchResponse>,
    210
        /// Request timestamp for latency tracking
    211
        pub submitted_at: std::time::Instant,
    212
    }
    213
    214
    /// Response from batch processor (PARITY-052)
    215
    #[cfg(feature = "gpu")]
    216
    #[derive(Debug, Clone)]
    217
    pub struct ContinuousBatchResponse {
    218
        /// Generated token IDs (includes prompt)
    219
        pub token_ids: Vec<u32>,
    220
        /// Number of prompt tokens (to skip when decoding)
    221
        pub prompt_len: usize,
    222
        /// Whether request was processed in batch or single-request path
    223
        pub batched: bool,
    224
        /// Batch size when processed (1 for single-request)
    225
        pub batch_size: usize,
    226
        /// Processing latency in milliseconds
    227
        pub latency_ms: f64,
    228
    }
    229
    230
    #[cfg(feature = "gpu")]
    231
    impl ContinuousBatchResponse {
    232
        /// Create response for single-request path
    233
    4
        pub fn single(token_ids: Vec<u32>, prompt_len: usize, latency_ms: f64) -> Self {
    234
    4
            Self {
    235
    4
                token_ids,
    236
    4
                prompt_len,
    237
    4
                batched: false,
    238
    4
                batch_size: 1,
    239
    4
                latency_ms,
    240
    4
            }
    241
    4
        }
    242
    243
        /// Create response for batched path
    244
    2
        pub fn batched(
    245
    2
            token_ids: Vec<u32>,
    246
    2
            prompt_len: usize,
    247
    2
            batch_size: usize,
    248
    2
            latency_ms: f64,
    249
    2
        ) -> Self {
    250
    2
            Self {
    251
    2
                token_ids,
    252
    2
                prompt_len,
    253
    2
                batched: true,
    254
    2
                batch_size,
    255
    2
                latency_ms,
    256
    2
            }
    257
    2
        }
    258
    259
        /// Get generated tokens (excluding prompt)
    260
    6
        pub fn generated_tokens(&self) -> &[u32] {
    261
    6
            if self.token_ids.len() > self.prompt_len {
    262
    5
                &self.token_ids[self.prompt_len..]
    263
            } else {
    264
    1
                &[]
    265
            }
    266
    6
        }
    267
    }
    268
    269
    /// Batch queue statistics (PARITY-052)
    270
    #[derive(Debug, Clone, Default)]
    271
    #[cfg(feature = "gpu")]
    272
    pub struct BatchQueueStats {
    273
        /// Total requests queued
    274
        pub total_queued: u64,
    275
        /// Total batches processed
    276
        pub total_batches: u64,
    277
        /// Total requests processed via single-request path
    278
        pub total_single: u64,
    279
        /// Average batch size
    280
        pub avg_batch_size: f64,
    281
        /// Average queue wait time in milliseconds
    282
        pub avg_wait_ms: f64,
    283
    }
    284
    285
    // ==================== PARITY-053: Batch Processor Background Task ====================
    286
    //
    287
    // Background task that processes batched inference requests.
    288
    // Collects requests until batch is ready (size threshold or timeout), then processes.
    289
    //
    290
    // Flow:
    291
    //   1. Receive requests via mpsc channel
    292
    //   2. Accumulate until batch_size >= optimal_batch OR window_ms timeout
    293
    //   3. Process batch using model.generate_with_cache() for each request
    294
    //   4. Send results via oneshot channels
    295
    //
    296
    // Note: True batch inference (single forward pass for multiple requests) requires
    297
    // additional model infrastructure. This implementation processes requests in
    298
    // parallel within a batch window, which still improves throughput under load.
    299
    // ==================================================================================
    300
    301
    /// Result from batch processing
    302
    #[cfg(feature = "gpu")]
    303
    #[derive(Debug)]
    304
    pub struct BatchProcessResult {
    305
        /// Number of requests processed
    306
        pub requests_processed: usize,
    307
        /// Whether processed as batch or single
    308
        pub was_batched: bool,
    309
        /// Total processing time in milliseconds
    310
        pub total_time_ms: f64,
    311
        /// Average latency per request in milliseconds
    312
        pub avg_latency_ms: f64,
    313
    }
    314
    315
    /// Spawn the batch processor background task (PARITY-053)
    316
    ///
    317
    /// Returns the sender channel for submitting requests.
    318
    /// The receiver is consumed by the spawned task.
    319
    ///
    320
    /// # Arguments
    321
    /// * `model` - The cached model for inference
    322
    /// * `config` - Batch configuration
    323
    ///
    324
    /// # Returns
    325
    /// * Sender channel for batch requests
    326
    #[cfg(feature = "gpu")]
    327
    0
    pub fn spawn_batch_processor(
    328
    0
        model: std::sync::Arc<crate::gguf::OwnedQuantizedModelCachedSync>,
    329
    0
        config: BatchConfig,
    330
    0
    ) -> tokio::sync::mpsc::Sender<ContinuousBatchRequest> {
    331
    0
        let (tx, rx) = tokio::sync::mpsc::channel(config.queue_size);
    332
    333
        // Spawn the background processor task
    334
    0
        tokio::spawn(batch_processor_task(rx, model, config));
    335
    336
    0
        tx
    337
    0
    }
    338
    339
    /// Background task that processes batched requests (PARITY-053)
    340
    ///
    341
    /// This task runs continuously, collecting requests and processing them in batches.
    342
    /// It uses a timeout-based batching strategy:
    343
    /// - Process immediately if batch reaches optimal_batch size
    344
    /// - Process on timeout (window_ms) if batch has requests
    345
    /// - Fall back to single-request processing for very small batches
    346
    #[cfg(feature = "gpu")]
    347
    0
    async fn batch_processor_task(
    348
    0
        mut rx: tokio::sync::mpsc::Receiver<ContinuousBatchRequest>,
    349
    0
        model: std::sync::Arc<crate::gguf::OwnedQuantizedModelCachedSync>,
    350
    0
        config: BatchConfig,
    351
    0
    ) {
    352
        use std::time::{Duration, Instant};
    353
        use tokio::time::timeout;
    354
    355
    0
        let mut batch: Vec<ContinuousBatchRequest> = Vec::with_capacity(config.max_batch);
    356
    0
        let mut window_start = Instant::now();
    357
    358
        loop {
    359
            // Calculate remaining time in window
    360
    0
            let elapsed = window_start.elapsed();
    361
    0
            let remaining = Duration::from_millis(config.window_ms).saturating_sub(elapsed);
    362
    363
            // Try to receive with timeout
    364
    0
            match timeout(remaining, rx.recv()).await {
    365
    0
                Ok(Some(request)) => {
    366
    0
                    batch.push(request);
    367
    368
                    // Process immediately if we hit optimal batch size
    369
    0
                    if batch.len() >= config.optimal_batch {
    370
    0
                        process_batch(&model, &config, &mut batch).await;
    371
    0
                        window_start = Instant::now();
    372
    0
                    }
    373
                },
    374
                Ok(None) => {
    375
                    // Channel closed, process remaining and exit
    376
    0
                    if !batch.is_empty() {
    377
    0
                        process_batch(&model, &config, &mut batch).await;
    378
    0
                    }
    379
    0
                    break;
    380
                },
    381
                Err(_) => {
    382
                    // Timeout - process current batch if we have requests
    383
    0
                    if !batch.is_empty() {
    384
    0
                        process_batch(&model, &config, &mut batch).await;
    385
    0
                    }
    386
    0
                    window_start = Instant::now();
    387
                },
    388
            }
    389
        }
    390
    0
    }
    391
    392
    /// Process a batch of requests (PARITY-053)
    393
    ///
    394
    /// Processes all requests in the batch and sends results via their oneshot channels.
    395
    /// Uses tokio::spawn to process requests concurrently within the batch.
    396
    #[cfg(feature = "gpu")]
    397
    0
    async fn process_batch(
    398
    0
        model: &std::sync::Arc<crate::gguf::OwnedQuantizedModelCachedSync>,
    399
    0
        config: &BatchConfig,
    400
    0
        batch: &mut Vec<ContinuousBatchRequest>,
    401
    0
    ) {
    402
        use std::time::Instant;
    403
    404
    0
        if batch.is_empty() {
    405
    0
            return;
    406
    0
        }
    407
    408
    0
        let batch_size = batch.len();
    409
    0
        let batch_start = Instant::now();
    410
    411
        // PARITY-095: Use configurable GPU batch threshold
    412
        // GPU GEMM wins at batch >= gpu_threshold (default 32, from IMP-600 analysis)
    413
    0
        let gpu_threshold = config.gpu_threshold;
    414
    415
        // Use true GPU batch inference if batch is large enough and GPU cache is warm
    416
    0
        if batch_size >= gpu_threshold && model.is_gpu_cache_warm() {
    417
            // PARITY-094: True batch inference with GPU FFN
    418
            // Collect all prompts
    419
    0
            let prompts: Vec<Vec<u32>> = batch.iter().map(|r| r.prompt_tokens.clone()).collect();
    420
    421
            // Use first request's config (batch inference assumes similar parameters)
    422
    0
            let first = &batch[0];
    423
    0
            let gen_config = crate::gguf::QuantizedGenerateConfig {
    424
    0
                max_tokens: first.max_tokens,
    425
    0
                temperature: first.temperature,
    426
    0
                top_k: first.top_k,
    427
    0
                stop_tokens: Vec::new(),
    428
    0
            };
    429
    430
            // Run batch generation with GPU FFN (PARITY-021)
    431
    0
            let results = model.batch_generate_gpu(&prompts, &gen_config);
    432
    433
    0
            let total_latency_ms = batch_start.elapsed().as_secs_f64() * 1000.0;
    434
    0
            let per_request_latency_ms = total_latency_ms / batch_size as f64;
    435
    436
            // Send responses
    437
    0
            match results {
    438
    0
                Ok(all_token_ids) => {
    439
    0
                    for (request, token_ids) in batch.drain(..).zip(all_token_ids.into_iter()) {
    440
    0
                        let response = ContinuousBatchResponse {
    441
    0
                            token_ids,
    442
    0
                            prompt_len: request.prompt_tokens.len(),
    443
    0
                            batched: true,
    444
    0
                            batch_size,
    445
    0
                            latency_ms: per_request_latency_ms,
    446
    0
                        };
    447
    0
                        let _ = request.response_tx.send(response);
    448
    0
                    }
    449
                },
    450
                Err(_) => {
    451
                    // Fallback: return prompts unchanged on error
    452
    0
                    for request in batch.drain(..) {
    453
    0
                        let response = ContinuousBatchResponse {
    454
    0
                            token_ids: request.prompt_tokens.clone(),
    455
    0
                            prompt_len: request.prompt_tokens.len(),
    456
    0
                            batched: false,
    457
    0
                            batch_size,
    458
    0
                            latency_ms: per_request_latency_ms,
    459
    0
                        };
    460
    0
                        let _ = request.response_tx.send(response);
    461
    0
                    }
    462
                },
    463
            }
    464
        } else {
    465
            // Concurrent single-request processing (for small batches or no GPU cache)
    466
    0
            let mut handles = Vec::with_capacity(batch_size);
    467
    468
    0
            for request in batch.drain(..) {
    469
    0
                let model = model.clone();
    470
    0
                let handle = tokio::spawn(async move {
    471
    0
                    let start = Instant::now();
    472
    473
                    // Build generation config
    474
    0
                    let gen_config = crate::gguf::QuantizedGenerateConfig {
    475
    0
                        max_tokens: request.max_tokens,
    476
    0
                        temperature: request.temperature,
    477
    0
                        top_k: request.top_k,
    478
    0
                        stop_tokens: Vec::new(),
    479
    0
                    };
    480
    481
                    // Generate
    482
    0
                    let result = model.generate_with_cache(&request.prompt_tokens, &gen_config);
    483
    484
    0
                    let latency_ms = start.elapsed().as_secs_f64() * 1000.0;
    485
    486
                    // Send response
    487
    0
                    let response = match result {
    488
    0
                        Ok(token_ids) => ContinuousBatchResponse {
    489
    0
                            token_ids,
    490
    0
                            prompt_len: request.prompt_tokens.len(),
    491
    0
                            batched: false,
    492
    0
                            batch_size: 1,
    493
    0
                            latency_ms,
    494
    0
                        },
    495
    0
                        Err(_) => ContinuousBatchResponse {
    496
    0
                            token_ids: request.prompt_tokens.clone(),
    497
    0
                            prompt_len: request.prompt_tokens.len(),
    498
    0
                            batched: false,
    499
    0
                            batch_size: 1,
    500
    0
                            latency_ms,
    501
    0
                        },
    502
                    };
    503
    504
                    // Send response (ignore if receiver dropped)
    505
    0
                    let _ = request.response_tx.send(response);
    506
    0
                });
    507
    508
    0
                handles.push(handle);
    509
            }
    510
    511
            // Wait for all to complete
    512
    0
            for handle in handles {
    513
    0
                let _ = handle.await;
    514
            }
    515
        }
    516
    0
    }
    517
    518
    /// GPU warmup handler (PARITY-022)
    519
    /// POST /v1/gpu/warmup - Warmup GPU cache for batch inference
    520
    #[cfg(feature = "gpu")]
    521
    1
    pub async fn gpu_warmup_handler(
    522
    1
        State(state): State<AppState>,
    523
    1
    ) -> Result<Json<GpuWarmupResponse>, (StatusCode, Json<ErrorResponse>)> {
    524
    1
        if let Some(cached_model0) = state.cached_model() {
    525
    0
            match cached_model.warmup_gpu_cache() {
    526
    0
                Ok((memory_bytes, num_layers)) => Ok(Json(GpuWarmupResponse {
    527
    0
                    success: true,
    528
    0
                    memory_bytes,
    529
    0
                    num_layers,
    530
    0
                    message: format!(
    531
    0
                        "GPU cache warmed up: {} layers, {:.2} GB",
    532
    0
                        num_layers,
    533
    0
                        memory_bytes as f64 / 1e9
    534
    0
                    ),
    535
    0
                })),
    536
    0
                Err(e) => Err((
    537
    0
                    StatusCode::INTERNAL_SERVER_ERROR,
    538
    0
                    Json(ErrorResponse {
    539
    0
                        error: format!("GPU warmup failed: {e}"),
    540
    0
                    }),
    541
    0
                )),
    542
            }
    543
        } else {
    544
    1
            Err((
    545
    1
                StatusCode::SERVICE_UNAVAILABLE,
    546
    1
                Json(ErrorResponse {
    547
    1
                    error: "No GPU-capable model loaded. Use with_cached_model() to enable."
    548
    1
                        .to_string(),
    549
    1
                }),
    550
    1
            ))
    551
        }
    552
    1
    }
    553
    554
    /// GPU warmup handler stub for non-GPU builds
    555
    #[cfg(not(feature = "gpu"))]
    556
    pub async fn gpu_warmup_handler(
    557
        State(_state): State<AppState>,
    558
    ) -> Result<Json<GpuWarmupResponse>, (StatusCode, Json<ErrorResponse>)> {
    559
        Err((
    560
            StatusCode::SERVICE_UNAVAILABLE,
    561
            Json(ErrorResponse {
    562
                error: "GPU feature not enabled. Build with --features gpu".to_string(),
    563
            }),
    564
        ))
    565
    }
    566
    567
    /// GPU status handler (PARITY-022)
    568
    /// GET /v1/gpu/status - Check GPU cache status
    569
    #[cfg(feature = "gpu")]
    570
    1
    pub async fn gpu_status_handler(
    571
    1
        State(state): State<AppState>,
    572
    1
    ) -> Result<Json<GpuStatusResponse>, (StatusCode, Json<ErrorResponse>)> {
    573
    1
        if let Some(cached_model0) = state.cached_model() {
    574
    0
            Ok(Json(GpuStatusResponse {
    575
    0
                cache_ready: cached_model.is_gpu_cache_warm(),
    576
    0
                cache_memory_bytes: cached_model.gpu_cache_memory(),
    577
    0
                batch_threshold: 32, // GPU GEMM threshold from IMP-600
    578
    0
                recommended_min_batch: 32,
    579
    0
            }))
    580
        } else {
    581
    1
            Ok(Json(GpuStatusResponse {
    582
    1
                cache_ready: false,
    583
    1
                cache_memory_bytes: 0,
    584
    1
                batch_threshold: 32,
    585
    1
                recommended_min_batch: 32,
    586
    1
            }))
    587
        }
    588
    1
    }
    589
    590
    /// GPU status handler stub for non-GPU builds
    591
    #[cfg(not(feature = "gpu"))]
    592
    pub async fn gpu_status_handler(
    593
        State(_state): State<AppState>,
    594
    ) -> Result<Json<GpuStatusResponse>, (StatusCode, Json<ErrorResponse>)> {
    595
        Ok(Json(GpuStatusResponse {
    596
            cache_ready: false,
    597
            cache_memory_bytes: 0,
    598
            batch_threshold: 32,
    599
            recommended_min_batch: 32,
    600
        }))
    601
    }
    602
    603
    /// GPU batch completions handler (PARITY-022)
    604
    /// POST /v1/batch/completions - GPU-accelerated batch inference
    605
    #[cfg(feature = "gpu")]
    606
    1
    pub async fn gpu_batch_completions_handler(
    607
    1
        State(state): State<AppState>,
    608
    1
        Json(request): Json<GpuBatchRequest>,
    609
    1
    ) -> Result<Json<GpuBatchResponse>, (StatusCode, Json<ErrorResponse>)> {
    610
        use std::time::Instant;
    611
    612
    1
        if request.prompts.is_empty() {
    613
    1
            return Err((
    614
    1
                StatusCode::BAD_REQUEST,
    615
    1
                Json(ErrorResponse {
    616
    1
                    error: "Prompts array cannot be empty".to_string(),
    617
    1
                }),
    618
    1
            ));
    619
    0
        }
    620
    621
    0
        let Some(cached_model) = state.cached_model() else {
    622
    0
            return Err((
    623
    0
                StatusCode::SERVICE_UNAVAILABLE,
    624
    0
                Json(ErrorResponse {
    625
    0
                    error: "No GPU-capable model loaded".to_string(),
    626
    0
                }),
    627
    0
            ));
    628
        };
    629
    630
        // Check if GPU cache is ready
    631
    0
        let gpu_ready = cached_model.is_gpu_cache_warm();
    632
    0
        let batch_size = request.prompts.len();
    633
    634
        // Tokenize all prompts
    635
        // For GPU batch, we need token IDs as Vec<Vec<u32>>
    636
    0
        let prompts_tokens: Vec<Vec<u32>> = request
    637
    0
            .prompts
    638
    0
            .iter()
    639
    0
            .map(|p| {
    640
                // Simple tokenization for batch - uses model's vocab
    641
                // In production, use a proper tokenizer
    642
    0
                p.bytes().map(|b| b as u32).collect()
    643
    0
            })
    644
    0
            .collect();
    645
    646
        // Create generation config
    647
    0
        let gen_config = crate::gguf::QuantizedGenerateConfig {
    648
    0
            max_tokens: request.max_tokens,
    649
    0
            temperature: request.temperature,
    650
    0
            top_k: request.top_k,
    651
    0
            stop_tokens: vec![],
    652
    0
        };
    653
    654
    0
        let start = Instant::now();
    655
    656
        // Decide GPU vs CPU path based on cache readiness and batch size
    657
    0
        let gpu_threshold = 32;
    658
    0
        let use_gpu = gpu_ready && batch_size >= gpu_threshold;
    659
    660
    0
        let results = if use_gpu {
    661
            // GPU batch inference path
    662
    0
            match cached_model.batch_generate_gpu(&prompts_tokens, &gen_config) {
    663
    0
                Ok(generated) => generated,
    664
    0
                Err(e) => {
    665
    0
                    return Err((
    666
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    667
    0
                        Json(ErrorResponse {
    668
    0
                            error: format!("GPU batch generation failed: {e}"),
    669
    0
                        }),
    670
    0
                    ));
    671
                },
    672
            }
    673
        } else {
    674
            // CPU sequential path (fallback)
    675
    0
            let mut results = Vec::with_capacity(batch_size);
    676
    0
            for prompt in &prompts_tokens {
    677
    0
                match cached_model.generate_with_cache(prompt, &gen_config) {
    678
    0
                    Ok(tokens) => results.push(tokens),
    679
    0
                    Err(e) => {
    680
    0
                        return Err((
    681
    0
                            StatusCode::INTERNAL_SERVER_ERROR,
    682
    0
                            Json(ErrorResponse {
    683
    0
                                error: format!("Generation failed: {e}"),
    684
    0
                            }),
    685
    0
                        ));
    686
                    },
    687
                }
    688
            }
    689
    0
            results
    690
        };
    691
    692
    0
        let elapsed = start.elapsed();
    693
    0
        let total_tokens: usize = results.iter().map(Vec::len).sum();
    694
    0
        let throughput_tps = total_tokens as f64 / elapsed.as_secs_f64();
    695
    696
        // Build response
    697
    0
        let batch_results: Vec<GpuBatchResult> = results
    698
    0
            .into_iter()
    699
    0
            .enumerate()
    700
    0
            .map(|(idx, tokens)| {
    701
    0
                let prompt_len = prompts_tokens.get(idx).map_or(0, Vec::len);
    702
    0
                let num_generated = tokens.len().saturating_sub(prompt_len);
    703
                GpuBatchResult {
    704
    0
                    index: idx,
    705
    0
                    token_ids: tokens.clone(),
    706
    0
                    text: tokens.iter().map(|&t| t as u8 as char).collect(),
    707
    0
                    num_generated,
    708
                }
    709
    0
            })
    710
    0
            .collect();
    711
    712
    0
        Ok(Json(GpuBatchResponse {
    713
    0
            results: batch_results,
    714
    0
            stats: GpuBatchStats {
    715
    0
                batch_size,
    716
    0
                gpu_used: use_gpu,
    717
    0
                total_tokens,
    718
    0
                processing_time_ms: elapsed.as_secs_f64() * 1000.0,
    719
    0
                throughput_tps,
    720
    0
            },
    721
    0
        }))
    722
    1
    }
    723
    724
    /// GPU batch completions handler stub for non-GPU builds
    725
    #[cfg(not(feature = "gpu"))]
    726
    pub async fn gpu_batch_completions_handler(
    727
        State(_state): State<AppState>,
    728
        Json(_request): Json<GpuBatchRequest>,
    729
    ) -> Result<Json<GpuBatchResponse>, (StatusCode, Json<ErrorResponse>)> {
    730
        Err((
    731
            StatusCode::SERVICE_UNAVAILABLE,
    732
            Json(ErrorResponse {
    733
                error: "GPU feature not enabled. Build with --features gpu".to_string(),
    734
            }),
    735
        ))
    736
    }
    737
    738
    /// Models list handler - returns available models in multi-model mode
    739
    1
    pub async fn models_handler(
    740
    1
        State(state): State<AppState>,
    741
    1
    ) -> Result<Json<ModelsResponse>, (StatusCode, Json<ErrorResponse>)> {
    742
    1
        if let Some(registry0) = &state.registry {
    743
    0
            let models = registry.list();
    744
    0
            Ok(Json(ModelsResponse { models }))
    745
        } else {
    746
            // Single model mode - return the single model info
    747
    1
            Ok(Json(ModelsResponse {
    748
    1
                models: vec![ModelInfo {
    749
    1
                    id: "default".to_string(),
    750
    1
                    name: "Default Model".to_string(),
    751
    1
                    description: "Single model deployment".to_string(),
    752
    1
                    format: "unknown".to_string(),
    753
    1
                    loaded: true,
    754
    1
                }],
    755
    1
            }))
    756
        }
    757
    1
    }
    758
    759
    /// Tokenize text handler
    760
    3
    pub async fn tokenize_handler(
    761
    3
        State(state): State<AppState>,
    762
    3
        Json(request): Json<TokenizeRequest>,
    763
    3
    ) -> Result<Json<TokenizeResponse>, (StatusCode, Json<ErrorResponse>)> {
    764
    3
        let (_model, tokenizer) = state.get_model(request.model_id.as_deref()).map_err(|e| {0
    765
    0
            (
    766
    0
                StatusCode::NOT_FOUND,
    767
    0
                Json(ErrorResponse {
    768
    0
                    error: e.to_string(),
    769
    0
                }),
    770
    0
            )
    771
    0
        })?;
    772
    773
    3
        let token_ids = tokenizer.encode(&request.text);
    774
    3
        let num_tokens = token_ids.len();
    775
    776
    3
        Ok(Json(TokenizeResponse {
    777
    3
            token_ids,
    778
    3
            num_tokens,
    779
    3
        }))
    780
    3
    }
    781
    782
    /// Generate text handler
    783
    12
    pub async fn generate_handler(
    784
    12
        State(state): State<AppState>,
    785
    12
        Json(request): Json<GenerateRequest>,
    786
    12
    ) -> Result<Json<GenerateResponse>, (StatusCode, Json<ErrorResponse>)> {
    787
        use std::time::Instant;
    788
    12
        let start = Instant::now();
    789
    790
        // Get model and tokenizer
    791
    12
        let (model, tokenizer) = state.get_model(request.model_id.as_deref()).map_err(|e| {0
    792
    0
            state.metrics.record_failure();
    793
    0
            (
    794
    0
                StatusCode::NOT_FOUND,
    795
    0
                Json(ErrorResponse {
    796
    0
                    error: e.to_string(),
    797
    0
                }),
    798
    0
            )
    799
    0
        })?;
    800
    801
        // Tokenize prompt
    802
    12
        let prompt_ids = tokenizer.encode(&request.prompt);
    803
    12
        if prompt_ids.is_empty() {
    804
    1
            state.metrics.record_failure();
    805
    1
            return Err((
    806
    1
                StatusCode::BAD_REQUEST,
    807
    1
                Json(ErrorResponse {
    808
    1
                    error: "Prompt cannot be empty".to_string(),
    809
    1
                }),
    810
    1
            ));
    811
    11
        }
    812
    813
        // Convert to usize for model
    814
    25
        let prompt11: Vec<usize>11 = prompt_ids.iter()11.map11(|&id| id as usize).collect11();
    815
    816
        // Build generation config
    817
    11
        let strategy9 = match request.strategy.as_str() {
    818
    11
            "greedy" => SamplingStrategy::Greedy5,
    819
    6
            "top_k" => SamplingStrategy::TopK { k: request.top_k }2,
    820
    4
            "top_p" => SamplingStrategy::TopP { p: request.top_p }2,
    821
            _ => {
    822
    2
                state.metrics.record_failure();
    823
    2
                return Err((
    824
    2
                    StatusCode::BAD_REQUEST,
    825
    2
                    Json(ErrorResponse {
    826
    2
                        error: format!("Invalid strategy: {}", request.strategy),
    827
    2
                    }),
    828
    2
                ));
    829
            },
    830
        };
    831
    832
    9
        let mut config = GenerationConfig::default()
    833
    9
            .with_max_tokens(request.max_tokens)
    834
    9
            .with_temperature(request.temperature);
    835
    836
    9
        config.strategy = strategy;
    837
    9
        if let Some(seed5) = request.seed {
    838
    5
            config = config.with_seed(seed);
    839
    5
        }4
    840
    841
        // Generate
    842
    9
        let generated = model.generate(&prompt, &config).map_err(|e| {0
    843
    0
            state.metrics.record_failure();
    844
    0
            (
    845
    0
                StatusCode::INTERNAL_SERVER_ERROR,
    846
    0
                Json(ErrorResponse {
    847
    0
                    error: e.to_string(),
    848
    0
                }),
    849
    0
            )
    850
    0
        })?;
    851
    852
        // Convert back to u32 and decode, with proper overflow handling
    853
    9
        let token_ids: Vec<u32> = generated
    854
    9
            .iter()
    855
    92
            .map9(|&id| {
    856
    92
                u32::try_from(id).map_err(|_| {0
    857
    0
                    (
    858
    0
                        StatusCode::BAD_REQUEST,
    859
    0
                        Json(ErrorResponse {
    860
    0
                            error: format!("Token ID {id} exceeds u32 range"),
    861
    0
                        }),
    862
    0
                    )
    863
    0
                })
    864
    92
            })
    865
    9
            .collect::<Result<Vec<_>, _>>()?0;
    866
    9
        let text = tokenizer.decode(&token_ids).map_err(|e| {0
    867
    0
            state.metrics.record_failure();
    868
    0
            (
    869
    0
                StatusCode::INTERNAL_SERVER_ERROR,
    870
    0
                Json(ErrorResponse {
    871
    0
                    error: e.to_string(),
    872
    0
                }),
    873
    0
            )
    874
    0
        })?;
    875
    876
    9
        let num_generated = generated.len() - prompt.len();
    877
    9
        let duration = start.elapsed();
    878
    879
        // Record successful generation with metrics
    880
    9
        state.metrics.record_success(num_generated, duration);
    881
    882
    9
        Ok(Json(GenerateResponse {
    883
    9
            token_ids,
    884
    9
            text,
    885
    9
            num_generated,
    886
    9
        }))
    887
    12
    }
    888
    889
    /// Batch tokenize handler
    890
    5
    pub async fn batch_tokenize_handler(
    891
    5
        State(state): State<AppState>,
    892
    5
        Json(request): Json<BatchTokenizeRequest>,
    893
    5
    ) -> Result<Json<BatchTokenizeResponse>, (StatusCode, Json<ErrorResponse>)> {
    894
    5
        if request.texts.is_empty() {
    895
    2
            return Err((
    896
    2
                StatusCode::BAD_REQUEST,
    897
    2
                Json(ErrorResponse {
    898
    2
                    error: "Texts array cannot be empty".to_string(),
    899
    2
                }),
    900
    2
            ));
    901
    3
        }
    902
    903
        // Get tokenizer (use default model)
    904
    3
        let (_model, tokenizer) = state.get_model(None).map_err(|e| {0
    905
    0
            (
    906
    0
                StatusCode::INTERNAL_SERVER_ERROR,
    907
    0
                Json(ErrorResponse {
    908
    0
                    error: e.to_string(),
    909
    0
                }),
    910
    0
            )
    911
    0
        })?;
    912
    913
        // Tokenize all texts
    914
    3
        let results: Vec<TokenizeResponse> = request
    915
    3
            .texts
    916
    3
            .iter()
    917
    9
            .map3(|text| {
    918
    9
                let token_ids = tokenizer.encode(text);
    919
    9
                let num_tokens = token_ids.len();
    920
    9
                TokenizeResponse {
    921
    9
                    token_ids,
    922
    9
                    num_tokens,
    923
    9
                }
    924
    9
            })
    925
    3
            .collect();
    926
    927
    3
        Ok(Json(BatchTokenizeResponse { results }))
    928
    5
    }
    929
    930
    /// Batch generate handler
    931
    11
    pub async fn batch_generate_handler(
    932
    11
        State(state): State<AppState>,
    933
    11
        Json(request): Json<BatchGenerateRequest>,
    934
    11
    ) -> Result<Json<BatchGenerateResponse>, (StatusCode, Json<ErrorResponse>)> {
    935
    11
        if request.prompts.is_empty() {
    936
    2
            return Err((
    937
    2
                StatusCode::BAD_REQUEST,
    938
    2
                Json(ErrorResponse {
    939
    2
                    error: "Prompts array cannot be empty".to_string(),
    940
    2
                }),
    941
    2
            ));
    942
    9
        }
    943
    944
        // Get model and tokenizer (use default model)
    945
    9
        let (model, tokenizer) = state.get_model(None).map_err(|e| {0
    946
    0
            (
    947
    0
                StatusCode::INTERNAL_SERVER_ERROR,
    948
    0
                Json(ErrorResponse {
    949
    0
                    error: e.to_string(),
    950
    0
                }),
    951
    0
            )
    952
    0
        })?;
    953
    954
        // Build generation config (shared across all prompts)
    955
    9
        let strategy7 = match request.strategy.as_str() {
    956
    9
            "greedy" => SamplingStrategy::Greedy5,
    957
    4
            "top_k" => SamplingStrategy::TopK { k: request.top_k }1,
    958
    3
            "top_p" => SamplingStrategy::TopP { p: request.top_p }1,
    959
            _ => {
    960
    2
                return Err((
    961
    2
                    StatusCode::BAD_REQUEST,
    962
    2
                    Json(ErrorResponse {
    963
    2
                        error: format!("Invalid strategy: {}", request.strategy),
    964
    2
                    }),
    965
    2
                ));
    966
            },
    967
        };
    968
    969
    7
        let mut config = GenerationConfig::default()
    970
    7
            .with_max_tokens(request.max_tokens)
    971
    7
            .with_temperature(request.temperature);
    972
    973
    7
        config.strategy = strategy;
    974
    7
        if let Some(seed4) = request.seed {
    975
    4
            config = config.with_seed(seed);
    976
    4
        }3
    977
    978
        // Process each prompt
    979
    7
        let mut results = Vec::with_capacity(request.prompts.len());
    980
    981
    19
        for prompt_text12 in &request.prompts {
    982
            // Tokenize prompt
    983
    12
            let prompt_ids = tokenizer.encode(prompt_text);
    984
    12
            if prompt_ids.is_empty() {
    985
    0
                return Err((
    986
    0
                    StatusCode::BAD_REQUEST,
    987
    0
                    Json(ErrorResponse {
    988
    0
                        error: format!("Prompt '{prompt_text}' tokenizes to empty sequence"),
    989
    0
                    }),
    990
    0
                ));
    991
    12
            }
    992
    993
            // Convert to usize for model
    994
    27
            let prompt12: Vec<usize>12 = prompt_ids.iter()12.map12(|&id| id as usize).collect12();
    995
    996
            // Generate
    997
    12
            let generated = model.generate(&prompt, &config).map_err(|e| {0
    998
    0
                (
    999
    0
                    StatusCode::INTERNAL_SERVER_ERROR,
    1000
    0
                    Json(ErrorResponse {
    1001
    0
                        error: e.to_string(),
    1002
    0
                    }),
    1003
    0
                )
    1004
    0
            })?;
    1005
    1006
            // Convert back to u32 and decode, with proper overflow handling
    1007
    12
            let token_ids: Vec<u32> = generated
    1008
    12
                .iter()
    1009
    153
                .map12(|&id| {
    1010
    153
                    u32::try_from(id).map_err(|_| {0
    1011
    0
                        (
    1012
    0
                            StatusCode::BAD_REQUEST,
    1013
    0
                            Json(ErrorResponse {
    1014
    0
                                error: format!("Token ID {id} exceeds u32 range"),
    1015
    0
                            }),
    1016
    0
                        )
    1017
    0
                    })
    1018
    153
                })
    1019
    12
                .collect::<Result<Vec<_>, _>>()?0;
    1020
    12
            let text = tokenizer.decode(&token_ids).map_err(|e| {0
    1021
    0
                (
    1022
    0
                    StatusCode::INTERNAL_SERVER_ERROR,
    1023
    0
                    Json(ErrorResponse {
    1024
    0
                        error: e.to_string(),
    1025
    0
                    }),
    1026
    0
                )
    1027
    0
            })?;
    1028
    1029
    12
            let num_generated = generated.len() - prompt.len();
    1030
    1031
    12
            results.push(GenerateResponse {
    1032
    12
                token_ids,
    1033
    12
                text,
    1034
    12
                num_generated,
    1035
    12
            });
    1036
        }
    1037
    1038
    7
        Ok(Json(BatchGenerateResponse { results }))
    1039
    11
    }
    1040
    1041
    /// Stream generate handler - generates tokens one by one via Server-Sent Events
    1042
    6
    pub async fn stream_generate_handler(
    1043
    6
        State(state): State<AppState>,
    1044
    6
        Json(request): Json<GenerateRequest>,
    1045
    6
    ) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, Json<ErrorResponse>)> {
    1046
        // Get model and tokenizer
    1047
    6
        let (model, tokenizer) = state.get_model(request.model_id.as_deref()).map_err(|e| {0
    1048
    0
            (
    1049
    0
                StatusCode::NOT_FOUND,
    1050
    0
                Json(ErrorResponse {
    1051
    0
                    error: e.to_string(),
    1052
    0
                }),
    1053
    0
            )
    1054
    0
        })?;
    1055
    1056
        // Tokenize prompt
    1057
    6
        let prompt_ids = tokenizer.encode(&request.prompt);
    1058
    6
        if prompt_ids.is_empty() {
    1059
    1
            return Err((
    1060
    1
                StatusCode::BAD_REQUEST,
    1061
    1
                Json(ErrorResponse {
    1062
    1
                    error: "Prompt cannot be empty".to_string(),
    1063
    1
                }),
    1064
    1
            ));
    1065
    5
        }
    1066
    1067
        // Convert to usize for model
    1068
    12
        let prompt5: Vec<usize>5 = prompt_ids.iter()5.map5(|&id| id as usize).collect5();
    1069
    5
        let prompt_len = prompt.len();
    1070
    1071
        // Build generation config
    1072
    5
        let strategy4 = match request.strategy.as_str() {
    1073
    5
            "greedy" => SamplingStrategy::Greedy2,
    1074
    3
            "top_k" => SamplingStrategy::TopK { k: request.top_k }1,
    1075
    2
            "top_p" => SamplingStrategy::TopP { p: request.top_p }1,
    1076
            _ => {
    1077
    1
                return Err((
    1078
    1
                    StatusCode::BAD_REQUEST,
    1079
    1
                    Json(ErrorResponse {
    1080
    1
                        error: format!("Invalid strategy: {}", request.strategy),
    1081
    1
                    }),
    1082
    1
                ));
    1083
            },
    1084
        };
    1085
    1086
    4
        let mut config = GenerationConfig::default()
    1087
    4
            .with_max_tokens(request.max_tokens)
    1088
    4
            .with_temperature(request.temperature);
    1089
    1090
    4
        config.strategy = strategy;
    1091
    4
        if let Some(seed2) = request.seed {
    1092
    2
            config = config.with_seed(seed);
    1093
    2
        }
    1094
    1095
        // Generate all tokens (in future, this will be truly streaming token-by-token)
    1096
    4
        let generated = match model.generate(&prompt, &config) {
    1097
    4
            Ok(tokens) => tokens,
    1098
    0
            Err(e) => {
    1099
    0
                return Err((
    1100
    0
                    StatusCode::INTERNAL_SERVER_ERROR,
    1101
    0
                    Json(ErrorResponse {
    1102
    0
                        error: e.to_string(),
    1103
    0
                    }),
    1104
    0
                ));
    1105
            },
    1106
        };
    1107
    1108
        // Convert to u32 with proper overflow handling
    1109
    4
        let token_ids: Vec<u32> = generated
    1110
    4
            .iter()
    1111
    16
            .map4(|&id| {
    1112
    16
                u32::try_from(id).map_err(|_| {0
    1113
    0
                    (
    1114
    0
                        StatusCode::BAD_REQUEST,
    1115
    0
                        Json(ErrorResponse {
    1116
    0
                            error: format!("Token ID {id} exceeds u32 range"),
    1117
    0
                        }),
    1118
    0
                    )
    1119
    0
                })
    1120
    16
            })
    1121
    4
            .collect::<Result<Vec<_>, _>>()?0;
    1122
    1123
        // Create stream that emits tokens one by one
    1124
    4
        let tokenizer_clone = tokenizer;
    1125
    4
        let stream = async_stream::stream! {
    1126
            // Skip prompt tokens, only stream generated tokens
    1127
            for &token_id in &token_ids[prompt_len..] {
    1128
                // Decode single token
    1129
                let text = match tokenizer_clone.decode(&[token_id]) {
    1130
                    Ok(t) => t,
    1131
                    Err(_) => String::from("<error>"),
    1132
                };
    1133
    1134
                let event = StreamTokenEvent { token_id, text };
    1135
                // Serialization of simple struct should not fail, but handle gracefully
    1136
                let data = serde_json::to_string(&event)
    1137
    0
                    .unwrap_or_else(|_| r#"{"error":"serialization failed"}"#.to_string());
    1138
    1139
                yield Ok::<_, Infallible>(Event::default().event("token").data(data));
    1140
            }
    1141
    1142
            // Send done event
    1143
            let done_event = StreamDoneEvent {
    1144
                num_generated: token_ids.len() - prompt_len,
    1145
            };
    1146
            // Serialization of simple struct should not fail, but handle gracefully
    1147
            let data = serde_json::to_string(&done_event)
    1148
    0
                .unwrap_or_else(|_| r#"{"error":"serialization failed"}"#.to_string());
    1149
            yield Ok(Event::default().event("done").data(data));
    1150
        };
    1151
    1152
    4
        Ok(Sse::new(stream))
    1153
    6
    }