## 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/openai_handlers.rs

    Line
    Count
    Source
    1
    //! OpenAI-compatible API handlers
    2
    //!
    3
    //! Extracted from api/mod.rs (PMAT-802) to reduce module size.
    4
    //! Contains chat completion, streaming, and model list handlers.
    5
    6
    use std::convert::Infallible;
    7
    8
    use axum::{
    9
        extract::State,
    10
        http::{HeaderMap, StatusCode},
    11
        response::{
    12
            sse::{Event, Sse},
    13
            IntoResponse, Response,
    14
        },
    15
        Json,
    16
    };
    17
    use futures::stream::Stream;
    18
    19
    use super::{
    20
        AppState, ChatChoice, ChatCompletionChunk, ChatCompletionRequest,
    21
        ChatCompletionResponse, ChatMessage, ErrorResponse, OpenAIModel,
    22
        OpenAIModelsResponse, Usage, build_trace_data,
    23
        format_chat_messages, clean_chat_output,
    24
    };
    25
    use crate::generate::{GenerationConfig, SamplingStrategy};
    26
    27
    /// OpenAI-compatible models listing handler
    28
    ///
    29
    /// Returns available models in OpenAI API format (GET /v1/models).
    30
    3
    pub async fn openai_models_handler(State(state): State<AppState>) -> Json<OpenAIModelsResponse> {
    31
    3
        let models = if let Some(registry0) = &state.registry {
    32
    0
            registry
    33
    0
                .list()
    34
    0
                .into_iter()
    35
    0
                .map(|m| OpenAIModel {
    36
    0
                    id: m.id,
    37
    0
                    object: "model".to_string(),
    38
    0
                    created: std::time::SystemTime::now()
    39
    0
                        .duration_since(std::time::UNIX_EPOCH)
    40
    0
                        .map(|d| d.as_secs() as i64)
    41
    0
                        .unwrap_or(0),
    42
    0
                    owned_by: "realizar".to_string(),
    43
    0
                })
    44
    0
                .collect()
    45
        } else {
    46
            // Single model mode
    47
    3
            vec![OpenAIModel {
    48
    3
                id: "default".to_string(),
    49
    3
                object: "model".to_string(),
    50
    3
                created: std::time::SystemTime::now()
    51
    3
                    .duration_since(std::time::UNIX_EPOCH)
    52
    3
                    .map(|d| d.as_secs() as i64)
    53
    3
                    .unwrap_or(0),
    54
    3
                owned_by: "realizar".to_string(),
    55
            }]
    56
        };
    57
    58
    3
        Json(OpenAIModelsResponse {
    59
    3
            object: "list".to_string(),
    60
    3
            data: models,
    61
    3
        })
    62
    3
    }
    63
    64
    /// OpenAI-compatible /v1/chat/completions endpoint (supports streaming)
    65
    6
    pub async fn openai_chat_completions_handler(
    66
    6
        State(state): State<AppState>,
    67
    6
        headers: HeaderMap,
    68
    6
        Json(request): Json<ChatCompletionRequest>,
    69
    6
    ) -> Response {
    70
        use std::time::Instant;
    71
    6
        let start = Instant::now();
    72
    73
        // Parse X-Trace-Level header for debugging
    74
    6
        let trace_level = headers
    75
    6
            .get("X-Trace-Level")
    76
    6
            .and_then(|v| v0.to_str0().ok0())
    77
    6
            .map(str::to_lowercase);
    78
    79
        // Generate request ID
    80
    6
        let request_id = format!(
    81
    6
            "chatcmpl-q4k-{}",
    82
    6
            std::time::SystemTime::now()
    83
    6
                .duration_since(std::time::UNIX_EPOCH)
    84
    6
                .unwrap_or_default()
    85
    6
                .as_millis()
    86
        );
    87
    88
        // IMP-152: Try GPU model (non-batched --gpu mode)
    89
        #[cfg(feature = "gpu")]
    90
    6
        if let Some(gpu_model_lock0) = state.gpu_model() {
    91
            use crate::gpu::GpuGenerateConfig;
    92
    93
    0
            let tokenizer = match state.tokenizer.clone() {
    94
    0
                Some(t) => t,
    95
                None => {
    96
    0
                    state.metrics.record_failure();
    97
    0
                    return (
    98
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    99
    0
                        Json(ErrorResponse {
    100
    0
                            error: "No tokenizer available".to_string(),
    101
    0
                        }),
    102
    0
                    )
    103
    0
                        .into_response();
    104
                },
    105
            };
    106
    107
            // Convert chat messages to prompt using ChatML
    108
    0
            let prompt_text = format_chat_messages(&request.messages, Some("qwen"));
    109
    0
            let prompt_ids: Vec<usize> = tokenizer
    110
    0
                .encode(&prompt_text)
    111
    0
                .iter()
    112
    0
                .map(|&x| x as usize)
    113
    0
                .collect();
    114
    115
    0
            if prompt_ids.is_empty() {
    116
    0
                state.metrics.record_failure();
    117
    0
                return (
    118
    0
                    StatusCode::BAD_REQUEST,
    119
    0
                    Json(ErrorResponse {
    120
    0
                        error: "Messages cannot be empty".to_string(),
    121
    0
                    }),
    122
    0
                )
    123
    0
                    .into_response();
    124
    0
            }
    125
    126
    0
            let prompt_tokens = prompt_ids.len();
    127
    0
            let max_tokens = request.max_tokens.unwrap_or(256);
    128
    0
            let temperature = request.temperature.unwrap_or(0.7);
    129
    130
            // PMAT-088: Get EOS token ID for proper stop sequence (GPU path)
    131
    0
            let eos_token_id = tokenizer
    132
    0
                .get_token_id("<|im_end|>")
    133
    0
                .or_else(|| tokenizer.get_token_id("<|endoftext|>"))
    134
    0
                .unwrap_or(151645) as usize;
    135
    136
    0
            let gpu_config = GpuGenerateConfig {
    137
    0
                max_tokens,
    138
    0
                temperature,
    139
    0
                top_k: if temperature == 0.0 { 1 } else { 40 },
    140
    0
                stop_tokens: vec![eos_token_id],
    141
            };
    142
    143
            // Generate using GPU model
    144
    0
            let generated = {
    145
    0
                let mut model = match gpu_model_lock.write() {
    146
    0
                    Ok(m) => m,
    147
    0
                    Err(e) => {
    148
    0
                        state.metrics.record_failure();
    149
    0
                        return (
    150
    0
                            StatusCode::INTERNAL_SERVER_ERROR,
    151
    0
                            Json(ErrorResponse {
    152
    0
                                error: format!("GPU model lock error: {e}"),
    153
    0
                            }),
    154
    0
                        )
    155
    0
                            .into_response();
    156
                    },
    157
                };
    158
    0
                match model.generate(&prompt_ids, &gpu_config) {
    159
    0
                    Ok(g) => g,
    160
    0
                    Err(e) => {
    161
    0
                        state.metrics.record_failure();
    162
    0
                        return (
    163
    0
                            StatusCode::INTERNAL_SERVER_ERROR,
    164
    0
                            Json(ErrorResponse {
    165
    0
                                error: e.to_string(),
    166
    0
                            }),
    167
    0
                        )
    168
    0
                            .into_response();
    169
                    },
    170
                }
    171
            };
    172
    173
            // Skip prompt tokens, convert to u32
    174
    0
            let token_ids: Vec<u32> = generated
    175
    0
                .iter()
    176
    0
                .skip(prompt_tokens)
    177
    0
                .map(|&x| x as u32)
    178
    0
                .collect();
    179
    0
            let completion_tokens = token_ids.len();
    180
    181
            // Handle streaming vs non-streaming
    182
    0
            if request.stream {
    183
    0
                let model_name = request.model.clone();
    184
    0
                let request_id_clone = request_id.clone();
    185
    186
    0
                let stream = async_stream::stream! {
    187
                    // Send initial chunk with role
    188
                    let initial = ChatCompletionChunk::initial(&request_id_clone, &model_name);
    189
                    if let Ok(data) = serde_json::to_string(&initial) {
    190
                        yield Ok::<_, Infallible>(Event::default().data(data));
    191
                    }
    192
    193
                    // Stream tokens one by one
    194
                    for &token_id in &token_ids {
    195
                        if let Ok(text) = tokenizer.decode(&[token_id]) {
    196
                            if !text.is_empty() {
    197
                                let chunk = ChatCompletionChunk::content(&request_id_clone, &model_name, &text);
    198
                                if let Ok(data) = serde_json::to_string(&chunk) {
    199
                                    yield Ok(Event::default().data(data));
    200
                                }
    201
                            }
    202
                        }
    203
                    }
    204
    205
                    // Send final chunk with finish reason
    206
                    let done = ChatCompletionChunk::done(&request_id_clone, &model_name);
    207
                    if let Ok(data) = serde_json::to_string(&done) {
    208
                        yield Ok(Event::default().data(data));
    209
                    }
    210
    211
                    // Send [DONE] marker
    212
                    yield Ok(Event::default().data("[DONE]".to_string()));
    213
                };
    214
    215
    0
                state
    216
    0
                    .metrics
    217
    0
                    .record_success(completion_tokens, start.elapsed());
    218
    0
                return Sse::new(stream).into_response();
    219
    0
            }
    220
    221
            // Non-streaming response
    222
    0
            let text = match tokenizer.decode(&token_ids) {
    223
    0
                Ok(t) => t,
    224
    0
                Err(e) => {
    225
    0
                    state.metrics.record_failure();
    226
    0
                    return (
    227
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    228
    0
                        Json(ErrorResponse {
    229
    0
                            error: e.to_string(),
    230
    0
                        }),
    231
    0
                    )
    232
    0
                        .into_response();
    233
                },
    234
            };
    235
    236
            // PMAT-088: Clean output to prevent prompt injection
    237
    0
            let text = clean_chat_output(&text);
    238
    239
    0
            let latency = start.elapsed();
    240
    0
            state.metrics.record_success(completion_tokens, latency);
    241
    242
            // Build trace data based on X-Trace-Level header (GPU path)
    243
    0
            let (brick_trace, step_trace, layer_trace) = build_trace_data(
    244
    0
                trace_level.as_deref(),
    245
    0
                latency.as_micros() as u64,
    246
    0
                prompt_tokens,
    247
    0
                completion_tokens,
    248
    0
                28, // Default layer count for Qwen2 models
    249
    0
            );
    250
    251
            return Json(ChatCompletionResponse {
    252
    0
                id: request_id,
    253
    0
                object: "chat.completion".to_string(),
    254
    0
                created: std::time::SystemTime::now()
    255
    0
                    .duration_since(std::time::UNIX_EPOCH)
    256
    0
                    .unwrap_or_default()
    257
    0
                    .as_secs() as i64,
    258
    0
                model: request.model.clone(),
    259
    0
                choices: vec![ChatChoice {
    260
                    index: 0,
    261
    0
                    message: ChatMessage {
    262
    0
                        role: "assistant".to_string(),
    263
    0
                        content: text,
    264
    0
                        name: None,
    265
    0
                    },
    266
    0
                    finish_reason: if completion_tokens >= max_tokens {
    267
    0
                        "length".to_string()
    268
                    } else {
    269
    0
                        "stop".to_string()
    270
                    },
    271
                }],
    272
    0
                usage: Usage {
    273
    0
                    prompt_tokens,
    274
    0
                    completion_tokens,
    275
    0
                    total_tokens: prompt_tokens + completion_tokens,
    276
    0
                },
    277
    0
                brick_trace,
    278
    0
                step_trace,
    279
    0
                layer_trace,
    280
            })
    281
    0
            .into_response();
    282
    6
        }
    283
    284
        // IMP-151: Try cached model (GPU batched --gpu --batch mode)
    285
        #[cfg(feature = "gpu")]
    286
    6
        if let Some(cached_model0) = state.cached_model() {
    287
            use crate::gguf::QuantizedGenerateConfig;
    288
    289
    0
            let tokenizer = match state.tokenizer.clone() {
    290
    0
                Some(t) => t,
    291
                None => {
    292
    0
                    state.metrics.record_failure();
    293
    0
                    return (
    294
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    295
    0
                        Json(ErrorResponse {
    296
    0
                            error: "No tokenizer available".to_string(),
    297
    0
                        }),
    298
    0
                    )
    299
    0
                        .into_response();
    300
                },
    301
            };
    302
    303
            // Convert chat messages to prompt using ChatML (GGUF models are typically Qwen/ChatML)
    304
    0
            let prompt_text = format_chat_messages(&request.messages, Some("qwen"));
    305
    306
            // Tokenize prompt
    307
    0
            let prompt_ids = tokenizer.encode(&prompt_text);
    308
    0
            if prompt_ids.is_empty() {
    309
    0
                state.metrics.record_failure();
    310
    0
                return (
    311
    0
                    StatusCode::BAD_REQUEST,
    312
    0
                    Json(ErrorResponse {
    313
    0
                        error: "Messages cannot be empty".to_string(),
    314
    0
                    }),
    315
    0
                )
    316
    0
                    .into_response();
    317
    0
            }
    318
    319
    0
            let prompt_tokens = prompt_ids.len();
    320
    0
            let max_tokens = request.max_tokens.unwrap_or(256);
    321
    0
            let temperature = request.temperature.unwrap_or(0.7);
    322
    323
            // PMAT-088: Get EOS token ID for proper stop sequence
    324
    0
            let eos_token_id = tokenizer
    325
    0
                .get_token_id("<|im_end|>")
    326
    0
                .or_else(|| tokenizer.get_token_id("<|endoftext|>"))
    327
    0
                .unwrap_or(151645);
    328
    329
    0
            let q_config = QuantizedGenerateConfig {
    330
    0
                max_tokens,
    331
    0
                temperature,
    332
    0
                top_k: if temperature == 0.0 { 1 } else { 40 },
    333
    0
                stop_tokens: vec![eos_token_id],
    334
            };
    335
    336
    0
            let generated = match cached_model
    337
    0
                .model()
    338
    0
                .generate_with_cache(&prompt_ids, &q_config)
    339
            {
    340
    0
                Ok(g) => g,
    341
    0
                Err(e) => {
    342
    0
                    state.metrics.record_failure();
    343
    0
                    return (
    344
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    345
    0
                        Json(ErrorResponse {
    346
    0
                            error: e.to_string(),
    347
    0
                        }),
    348
    0
                    )
    349
    0
                        .into_response();
    350
                },
    351
            };
    352
    353
            // Skip prompt tokens
    354
    0
            let token_ids: Vec<u32> = generated.iter().skip(prompt_tokens).copied().collect();
    355
    0
            let completion_tokens = token_ids.len();
    356
    357
            // Handle streaming vs non-streaming
    358
    0
            if request.stream {
    359
                // Streaming response - return SSE
    360
    0
                let model_name = request.model.clone();
    361
    0
                let request_id_clone = request_id.clone();
    362
    363
    0
                let stream = async_stream::stream! {
    364
                    // Send initial chunk with role
    365
                    let initial = ChatCompletionChunk::initial(&request_id_clone, &model_name);
    366
                    if let Ok(data) = serde_json::to_string(&initial) {
    367
                        yield Ok::<_, Infallible>(Event::default().data(data));
    368
                    }
    369
    370
                    // Stream tokens one by one
    371
                    for &token_id in &token_ids {
    372
                        // Decode single token
    373
                        if let Ok(text) = tokenizer.decode(&[token_id]) {
    374
                            if !text.is_empty() {
    375
                                let chunk = ChatCompletionChunk::content(&request_id_clone, &model_name, &text);
    376
                                if let Ok(data) = serde_json::to_string(&chunk) {
    377
                                    yield Ok(Event::default().data(data));
    378
                                }
    379
                            }
    380
                        }
    381
                    }
    382
    383
                    // Send final chunk with finish reason
    384
                    let done = ChatCompletionChunk::done(&request_id_clone, &model_name);
    385
                    if let Ok(data) = serde_json::to_string(&done) {
    386
                        yield Ok(Event::default().data(data));
    387
                    }
    388
    389
                    // Send [DONE] marker
    390
                    yield Ok(Event::default().data("[DONE]".to_string()));
    391
                };
    392
    393
    0
                state
    394
    0
                    .metrics
    395
    0
                    .record_success(completion_tokens, start.elapsed());
    396
    0
                return Sse::new(stream).into_response();
    397
    0
            }
    398
    399
            // Non-streaming response
    400
    0
            let text = match tokenizer.decode(&token_ids) {
    401
    0
                Ok(t) => t,
    402
    0
                Err(e) => {
    403
    0
                    state.metrics.record_failure();
    404
    0
                    return (
    405
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    406
    0
                        Json(ErrorResponse {
    407
    0
                            error: e.to_string(),
    408
    0
                        }),
    409
    0
                    )
    410
    0
                        .into_response();
    411
                },
    412
            };
    413
    414
            // PMAT-088: Clean output to prevent prompt injection
    415
    0
            let text = clean_chat_output(&text);
    416
    417
    0
            let latency = start.elapsed();
    418
    0
            state.metrics.record_success(completion_tokens, latency);
    419
    420
            // Build trace data based on X-Trace-Level header (cached GPU path)
    421
    0
            let (brick_trace, step_trace, layer_trace) = build_trace_data(
    422
    0
                trace_level.as_deref(),
    423
    0
                latency.as_micros() as u64,
    424
    0
                prompt_tokens,
    425
    0
                completion_tokens,
    426
    0
                28, // Default layer count for Qwen2 models
    427
    0
            );
    428
    429
            return Json(ChatCompletionResponse {
    430
    0
                id: request_id,
    431
    0
                object: "chat.completion".to_string(),
    432
    0
                created: std::time::SystemTime::now()
    433
    0
                    .duration_since(std::time::UNIX_EPOCH)
    434
    0
                    .unwrap_or_default()
    435
    0
                    .as_secs() as i64,
    436
    0
                model: request.model.clone(),
    437
    0
                choices: vec![ChatChoice {
    438
                    index: 0,
    439
    0
                    message: ChatMessage {
    440
    0
                        role: "assistant".to_string(),
    441
    0
                        content: text,
    442
    0
                        name: None,
    443
    0
                    },
    444
    0
                    finish_reason: if completion_tokens >= max_tokens {
    445
    0
                        "length".to_string()
    446
                    } else {
    447
    0
                        "stop".to_string()
    448
                    },
    449
                }],
    450
    0
                usage: Usage {
    451
    0
                    prompt_tokens,
    452
    0
                    completion_tokens,
    453
    0
                    total_tokens: prompt_tokens + completion_tokens,
    454
    0
                },
    455
    0
                brick_trace,
    456
    0
                step_trace,
    457
    0
                layer_trace,
    458
            })
    459
    0
            .into_response();
    460
    6
        }
    461
    462
        // PAR-111: CUDA-optimized model for high-performance GPU inference (755+ tok/s, 2.6x Ollama)
    463
        #[cfg(feature = "cuda")]
    464
        if let Some(cuda_model_lock) = state.cuda_model() {
    465
            use crate::gguf::QuantizedGenerateConfig;
    466
    467
            let tokenizer = match state.tokenizer.clone() {
    468
                Some(t) => t,
    469
                None => {
    470
                    state.metrics.record_failure();
    471
                    return (
    472
                        StatusCode::INTERNAL_SERVER_ERROR,
    473
                        Json(ErrorResponse {
    474
                            error: "No tokenizer available".to_string(),
    475
                        }),
    476
                    )
    477
                        .into_response();
    478
                },
    479
            };
    480
    481
            // Convert chat messages to prompt using ChatML (GGUF models are typically Qwen/ChatML)
    482
            let prompt_text = format_chat_messages(&request.messages, Some("qwen"));
    483
    484
            // Tokenize prompt
    485
            let prompt_ids = tokenizer.encode(&prompt_text);
    486
            if prompt_ids.is_empty() {
    487
                state.metrics.record_failure();
    488
                return (
    489
                    StatusCode::BAD_REQUEST,
    490
                    Json(ErrorResponse {
    491
                        error: "Messages cannot be empty".to_string(),
    492
                    }),
    493
                )
    494
                    .into_response();
    495
            }
    496
    497
            let prompt_tokens = prompt_ids.len();
    498
            let max_tokens = request.max_tokens.unwrap_or(256);
    499
            let temperature = request.temperature.unwrap_or(0.7);
    500
    501
            // PMAT-088: Get EOS token ID for proper stop sequence
    502
            let eos_token_id = tokenizer
    503
                .get_token_id("<|im_end|>")
    504
                .or_else(|| tokenizer.get_token_id("<|endoftext|>"))
    505
                .unwrap_or(151645);
    506
    507
            let q_config = QuantizedGenerateConfig {
    508
                max_tokens,
    509
                temperature,
    510
                top_k: if temperature == 0.0 { 1 } else { 40 },
    511
                stop_tokens: vec![eos_token_id],
    512
            };
    513
    514
            // PAR-112: True streaming - handle streaming vs non-streaming with different paths
    515
            if request.stream {
    516
                // TRUE STREAMING: Generate tokens one-by-one and stream as they're produced
    517
                use tokio::sync::mpsc;
    518
                use tokio_stream::wrappers::ReceiverStream;
    519
                use tokio_stream::StreamExt;
    520
    521
                let (tx, rx) = mpsc::channel::<Result<u32, String>>(16);
    522
                let cuda_model_clone = cuda_model_lock.clone();
    523
                let prompt_ids_clone = prompt_ids.clone();
    524
                let q_config_clone = q_config.clone();
    525
    526
                // Spawn generation in a blocking task to avoid blocking the async runtime
    527
                tokio::task::spawn_blocking(move || {
    528
                    let mut cuda_model = cuda_model_clone.write().expect("operation failed");
    529
    530
                    // Use streaming generation - sends tokens via channel as they're generated
    531
                    let result = cuda_model.generate_gpu_resident_streaming(
    532
                        &prompt_ids_clone,
    533
                        &q_config_clone,
    534
                        |token_id| {
    535
                            // Send token through channel; return false to stop if channel closed
    536
                            tx.blocking_send(Ok(token_id)).is_ok()
    537
                        },
    538
                    );
    539
    540
                    // Send error if generation failed
    541
                    if let Err(e) = result {
    542
                        let _ = tx.blocking_send(Err(e.to_string()));
    543
                    }
    544
                });
    545
    546
                // Convert channel receiver to SSE stream
    547
                let model_name = request.model.clone();
    548
                let request_id_clone = request_id.clone();
    549
                let tokenizer_clone = tokenizer.clone();
    550
                let metrics = state.metrics.clone();
    551
                let start_time = start;
    552
    553
                let token_stream = ReceiverStream::new(rx);
    554
                let mut completion_tokens = 0usize;
    555
    556
                let stream = async_stream::stream! {
    557
                    // Send initial chunk with role
    558
                    let initial = ChatCompletionChunk::initial(&request_id_clone, &model_name);
    559
                    if let Ok(data) = serde_json::to_string(&initial) {
    560
                        yield Ok::<_, Infallible>(Event::default().data(data));
    561
                    }
    562
    563
                    // Stream tokens as they arrive from generation
    564
                    tokio::pin!(token_stream);
    565
                    while let Some(result) = token_stream.next().await {
    566
                        match result {
    567
                            Ok(token_id) => {
    568
                                completion_tokens += 1;
    569
                                // Decode and send immediately
    570
                                if let Ok(text) = tokenizer_clone.decode(&[token_id]) {
    571
                                    if !text.is_empty() {
    572
                                        let chunk = ChatCompletionChunk::content(&request_id_clone, &model_name, &text);
    573
                                        if let Ok(data) = serde_json::to_string(&chunk) {
    574
                                            yield Ok(Event::default().data(data));
    575
                                        }
    576
                                    }
    577
                                }
    578
                            }
    579
                            Err(e) => {
    580
                                // Send error chunk
    581
                                let error_chunk = serde_json::json!({
    582
                                    "error": e
    583
                                });
    584
                                if let Ok(data) = serde_json::to_string(&error_chunk) {
    585
                                    yield Ok(Event::default().data(data));
    586
                                }
    587
                                break;
    588
                            }
    589
                        }
    590
                    }
    591
    592
                    // Send final chunk with finish reason
    593
                    let done = ChatCompletionChunk::done(&request_id_clone, &model_name);
    594
                    if let Ok(data) = serde_json::to_string(&done) {
    595
                        yield Ok(Event::default().data(data));
    596
                    }
    597
    598
                    // Record metrics
    599
                    metrics.record_success(completion_tokens, start_time.elapsed());
    600
    601
                    // Send [DONE] marker
    602
                    yield Ok(Event::default().data("[DONE]"));
    603
                };
    604
    605
                return Sse::new(stream)
    606
                    .keep_alive(
    607
                        axum::response::sse::KeepAlive::new()
    608
                            .interval(std::time::Duration::from_secs(15))
    609
                            .text("keep-alive"),
    610
                    )
    611
                    .into_response();
    612
            }
    613
    614
            // NON-STREAMING: Generate all tokens first, then return
    615
            let mut cuda_model = cuda_model_lock.write().expect("operation failed");
    616
    617
            let generated = match cuda_model.generate_gpu_resident(&prompt_ids, &q_config) {
    618
                Ok(g) => g,
    619
                Err(e) => {
    620
                    state.metrics.record_failure();
    621
                    return (
    622
                        StatusCode::INTERNAL_SERVER_ERROR,
    623
                        Json(ErrorResponse {
    624
                            error: e.to_string(),
    625
                        }),
    626
                    )
    627
                        .into_response();
    628
                },
    629
            };
    630
    631
            // Skip prompt tokens
    632
            let token_ids: Vec<u32> = generated.iter().skip(prompt_tokens).copied().collect();
    633
            let completion_tokens = token_ids.len();
    634
    635
            // Non-streaming: decode all tokens and return
    636
            let response_text = tokenizer
    637
                .decode(&token_ids)
    638
                .unwrap_or_else(|_| String::new());
    639
    640
            // PMAT-088: Clean output to prevent prompt injection
    641
            let response_text = clean_chat_output(&response_text);
    642
    643
            let elapsed = start.elapsed();
    644
            state.metrics.record_success(completion_tokens, elapsed);
    645
    646
            // Build trace data based on X-Trace-Level header (CUDA optimized path)
    647
            let (brick_trace, step_trace, layer_trace) = build_trace_data(
    648
                trace_level.as_deref(),
    649
                elapsed.as_micros() as u64,
    650
                prompt_tokens,
    651
                completion_tokens,
    652
                28, // Default layer count for Qwen2 models
    653
            );
    654
    655
            return Json(ChatCompletionResponse {
    656
                id: request_id,
    657
                object: "chat.completion".to_string(),
    658
                created: std::time::SystemTime::now()
    659
                    .duration_since(std::time::UNIX_EPOCH)
    660
                    .unwrap_or_default()
    661
                    .as_secs() as i64,
    662
                model: request.model,
    663
                choices: vec![ChatChoice {
    664
                    index: 0,
    665
                    message: ChatMessage {
    666
                        role: "assistant".to_string(),
    667
                        content: response_text,
    668
                        name: None,
    669
                    },
    670
                    finish_reason: "stop".to_string(),
    671
                }],
    672
                usage: Usage {
    673
                    prompt_tokens,
    674
                    completion_tokens,
    675
                    total_tokens: prompt_tokens + completion_tokens,
    676
                },
    677
                brick_trace,
    678
                step_trace,
    679
                layer_trace,
    680
            })
    681
            .into_response();
    682
        }
    683
    684
        // IMP-150: Try quantized model (supports GGUF serve mode)
    685
    6
        if let Some(quantized_model0) = state.quantized_model() {
    686
            use crate::gguf::QuantizedGenerateConfig;
    687
    688
    0
            let tokenizer = match state.tokenizer.clone() {
    689
    0
                Some(t) => t,
    690
                None => {
    691
    0
                    state.metrics.record_failure();
    692
    0
                    return (
    693
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    694
    0
                        Json(ErrorResponse {
    695
    0
                            error: "No tokenizer available".to_string(),
    696
    0
                        }),
    697
    0
                    )
    698
    0
                        .into_response();
    699
                },
    700
            };
    701
    702
            // Convert chat messages to prompt using ChatML (GGUF models are typically Qwen/ChatML)
    703
    0
            let prompt_text = format_chat_messages(&request.messages, Some("qwen"));
    704
    705
            // Tokenize prompt
    706
    0
            let prompt_ids = tokenizer.encode(&prompt_text);
    707
    0
            if prompt_ids.is_empty() {
    708
    0
                state.metrics.record_failure();
    709
    0
                return (
    710
    0
                    StatusCode::BAD_REQUEST,
    711
    0
                    Json(ErrorResponse {
    712
    0
                        error: "Messages cannot be empty".to_string(),
    713
    0
                    }),
    714
    0
                )
    715
    0
                    .into_response();
    716
    0
            }
    717
    718
    0
            let prompt_tokens = prompt_ids.len();
    719
    0
            let max_tokens = request.max_tokens.unwrap_or(256);
    720
    0
            let temperature = request.temperature.unwrap_or(0.7);
    721
    722
            // PMAT-088: Get EOS token ID for proper stop sequence
    723
            // ChatML uses <|im_end|> (token ID 151645 for Qwen models)
    724
    0
            let eos_token_id = tokenizer
    725
    0
                .get_token_id("<|im_end|>")
    726
    0
                .or_else(|| tokenizer.get_token_id("<|endoftext|>"))
    727
    0
                .unwrap_or(151645); // Fallback to Qwen's <|im_end|> token ID
    728
    729
    0
            let q_config = QuantizedGenerateConfig {
    730
    0
                max_tokens,
    731
    0
                temperature,
    732
    0
                top_k: if temperature == 0.0 { 1 } else { 40 },
    733
    0
                stop_tokens: vec![eos_token_id],
    734
            };
    735
    736
            // PMAT-087: True streaming - handle streaming vs non-streaming with different paths
    737
    0
            if request.stream {
    738
                // TRUE STREAMING: Generate tokens one-by-one and stream as they're produced
    739
                use tokio::sync::mpsc;
    740
                use tokio_stream::wrappers::ReceiverStream;
    741
                use tokio_stream::StreamExt;
    742
    743
    0
                let (tx, rx) = mpsc::channel::<Result<u32, String>>(16);
    744
    0
                let quantized_model_clone = quantized_model.clone();
    745
    0
                let prompt_ids_clone = prompt_ids.clone();
    746
    0
                let q_config_clone = q_config.clone();
    747
    748
                // Spawn generation in a blocking task to avoid blocking the async runtime
    749
    0
                tokio::task::spawn_blocking(move || {
    750
                    // Use streaming generation - sends tokens via channel as they're generated
    751
    0
                    let result = quantized_model_clone.generate_with_cache_streaming(
    752
    0
                        &prompt_ids_clone,
    753
    0
                        &q_config_clone,
    754
    0
                        |token_id| {
    755
                            // Send token through channel; return false to stop if channel closed
    756
    0
                            tx.blocking_send(Ok(token_id)).is_ok()
    757
    0
                        },
    758
                    );
    759
    760
                    // Send error if generation failed
    761
    0
                    if let Err(e) = result {
    762
    0
                        let _ = tx.blocking_send(Err(e.to_string()));
    763
    0
                    }
    764
    0
                });
    765
    766
                // Convert channel receiver to SSE stream
    767
    0
                let model_name = request.model.clone();
    768
    0
                let request_id_clone = request_id.clone();
    769
    0
                let tokenizer_clone = tokenizer.clone();
    770
    0
                let metrics = state.metrics.clone();
    771
    0
                let start_time = start;
    772
    773
    0
                let token_stream = ReceiverStream::new(rx);
    774
    0
                let mut completion_tokens = 0usize;
    775
    776
    0
                let stream = async_stream::stream! {
    777
                    // Send initial chunk with role
    778
                    let initial = ChatCompletionChunk::initial(&request_id_clone, &model_name);
    779
                    if let Ok(data) = serde_json::to_string(&initial) {
    780
                        yield Ok::<_, Infallible>(Event::default().data(data));
    781
                    }
    782
    783
                    // Stream tokens as they arrive from generation
    784
    0
                    tokio::pin!(token_stream);
    785
                    while let Some(result) = token_stream.next().await {
    786
                        match result {
    787
                            Ok(token_id) => {
    788
                                completion_tokens += 1;
    789
                                // Decode and send immediately
    790
                                if let Ok(text) = tokenizer_clone.decode(&[token_id]) {
    791
                                    // PMAT-088: Clean individual tokens of stop sequences
    792
                                    let cleaned = clean_chat_output(&text);
    793
                                    if !cleaned.is_empty() {
    794
                                        let chunk = ChatCompletionChunk::content(&request_id_clone, &model_name, &cleaned);
    795
                                        if let Ok(data) = serde_json::to_string(&chunk) {
    796
                                            yield Ok(Event::default().data(data));
    797
                                        }
    798
                                    }
    799
                                }
    800
                            }
    801
                            Err(e) => {
    802
                                // Send error chunk
    803
                                let error_chunk = serde_json::json!({
    804
                                    "error": e
    805
                                });
    806
                                if let Ok(data) = serde_json::to_string(&error_chunk) {
    807
                                    yield Ok(Event::default().data(data));
    808
                                }
    809
                                break;
    810
                            }
    811
                        }
    812
                    }
    813
    814
                    // Send final chunk with finish reason
    815
                    let done = ChatCompletionChunk::done(&request_id_clone, &model_name);
    816
                    if let Ok(data) = serde_json::to_string(&done) {
    817
                        yield Ok(Event::default().data(data));
    818
                    }
    819
    820
                    // Record metrics
    821
                    metrics.record_success(completion_tokens, start_time.elapsed());
    822
    823
                    // Send [DONE] marker
    824
                    yield Ok(Event::default().data("[DONE]"));
    825
                };
    826
    827
    0
                return Sse::new(stream)
    828
    0
                    .keep_alive(
    829
    0
                        axum::response::sse::KeepAlive::new()
    830
    0
                            .interval(std::time::Duration::from_secs(15))
    831
    0
                            .text("keep-alive"),
    832
                    )
    833
    0
                    .into_response();
    834
    0
            }
    835
    836
            // NON-STREAMING: Generate all tokens first, then return
    837
    0
            let generated = match quantized_model.generate_with_cache(&prompt_ids, &q_config) {
    838
    0
                Ok(g) => g,
    839
    0
                Err(e) => {
    840
    0
                    state.metrics.record_failure();
    841
    0
                    return (
    842
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    843
    0
                        Json(ErrorResponse {
    844
    0
                            error: e.to_string(),
    845
    0
                        }),
    846
    0
                    )
    847
    0
                        .into_response();
    848
                },
    849
            };
    850
    851
            // Skip prompt tokens
    852
    0
            let token_ids: Vec<u32> = generated.iter().skip(prompt_tokens).copied().collect();
    853
    0
            let completion_tokens = token_ids.len();
    854
    855
            // Non-streaming response - return JSON
    856
    0
            let text = match tokenizer.decode(&token_ids) {
    857
    0
                Ok(t) => t,
    858
    0
                Err(e) => {
    859
    0
                    state.metrics.record_failure();
    860
    0
                    return (
    861
    0
                        StatusCode::INTERNAL_SERVER_ERROR,
    862
    0
                        Json(ErrorResponse {
    863
    0
                            error: e.to_string(),
    864
    0
                        }),
    865
    0
                    )
    866
    0
                        .into_response();
    867
                },
    868
            };
    869
    870
            // PMAT-088: Clean output - stop at first stop sequence to prevent prompt injection
    871
    0
            let text = clean_chat_output(&text);
    872
    873
    0
            let latency = start.elapsed();
    874
    0
            state.metrics.record_success(completion_tokens, latency);
    875
    876
            // Build trace data based on X-Trace-Level header (quantized model path)
    877
    0
            let (brick_trace, step_trace, layer_trace) = build_trace_data(
    878
    0
                trace_level.as_deref(),
    879
    0
                latency.as_micros() as u64,
    880
    0
                prompt_tokens,
    881
    0
                completion_tokens,
    882
    0
                28, // Default layer count for Qwen2 models
    883
    0
            );
    884
    885
            return Json(ChatCompletionResponse {
    886
    0
                id: request_id,
    887
    0
                object: "chat.completion".to_string(),
    888
    0
                created: std::time::SystemTime::now()
    889
    0
                    .duration_since(std::time::UNIX_EPOCH)
    890
    0
                    .unwrap_or_default()
    891
    0
                    .as_secs() as i64,
    892
    0
                model: request.model.clone(),
    893
    0
                choices: vec![ChatChoice {
    894
                    index: 0,
    895
    0
                    message: ChatMessage {
    896
    0
                        role: "assistant".to_string(),
    897
    0
                        content: text,
    898
    0
                        name: None,
    899
    0
                    },
    900
    0
                    finish_reason: if completion_tokens >= max_tokens {
    901
    0
                        "length".to_string()
    902
                    } else {
    903
    0
                        "stop".to_string()
    904
                    },
    905
                }],
    906
    0
                usage: Usage {
    907
    0
                    prompt_tokens,
    908
    0
                    completion_tokens,
    909
    0
                    total_tokens: prompt_tokens + completion_tokens,
    910
    0
                },
    911
    0
                brick_trace,
    912
    0
                step_trace,
    913
    0
                layer_trace,
    914
            })
    915
    0
            .into_response();
    916
    6
        }
    917
    918
        // Fall back to registry-based model lookup
    919
    6
        let model_id = if request.model == "default" || request.model1.is_empty1() {
    920
    5
            None
    921
        } else {
    922
    1
            Some(request.model.as_str())
    923
        };
    924
    925
    6
        let (model, tokenizer) = match state.get_model(model_id) {
    926
    6
            Ok((m, t)) => (m, t),
    927
    0
            Err(e) => {
    928
    0
                state.metrics.record_failure();
    929
    0
                return (
    930
    0
                    StatusCode::NOT_FOUND,
    931
    0
                    Json(ErrorResponse {
    932
    0
                        error: e.to_string(),
    933
    0
                    }),
    934
    0
                )
    935
    0
                    .into_response();
    936
            },
    937
        };
    938
    939
        // Convert chat messages to prompt using model-specific template
    940
    6
        let prompt_text = format_chat_messages(&request.messages, Some(&request.model));
    941
    942
        // Tokenize prompt
    943
    6
        let prompt_ids = tokenizer.encode(&prompt_text);
    944
    6
        if prompt_ids.is_empty() {
    945
    2
            state.metrics.record_failure();
    946
    2
            return (
    947
    2
                StatusCode::BAD_REQUEST,
    948
    2
                Json(ErrorResponse {
    949
    2
                    error: "Messages cannot be empty".to_string(),
    950
    2
                }),
    951
    2
            )
    952
    2
                .into_response();
    953
    4
        }
    954
    955
    4
        let prompt_tokens = prompt_ids.len();
    956
    957
        // Convert to usize for model
    958
    46
        let prompt4: Vec<usize>4 = prompt_ids.iter()4.map4(|&id| id as usize).collect4();
    959
    960
        // Build generation config
    961
    4
        let max_tokens = request.max_tokens.unwrap_or(256);
    962
    4
        let temperature = request.temperature.unwrap_or(0.7);
    963
    964
    4
        let mut config = GenerationConfig::default()
    965
    4
            .with_max_tokens(max_tokens)
    966
    4
            .with_temperature(temperature);
    967
    968
    4
        if let Some(top_p1) = request.top_p {
    969
    1
            config.strategy = SamplingStrategy::TopP { p: top_p };
    970
    3
        }
    971
    972
        // Generate
    973
    4
        let generated = match model.generate(&prompt, &config) {
    974
    4
            Ok(g) => g,
    975
    0
            Err(e) => {
    976
    0
                state.metrics.record_failure();
    977
    0
                return (
    978
    0
                    StatusCode::INTERNAL_SERVER_ERROR,
    979
    0
                    Json(ErrorResponse {
    980
    0
                        error: e.to_string(),
    981
    0
                    }),
    982
    0
                )
    983
    0
                    .into_response();
    984
            },
    985
        };
    986
    987
        // Convert back to u32 and decode
    988
    4
        let token_ids: Vec<u32> = match generated
    989
    4
            .iter()
    990
    573
            .map4(|&id| u32::try_from(id).map_err(|_| format!0("Token ID {id} exceeds u32 range"0)))
    991
    4
            .collect::<Result<Vec<_>, _>>()
    992
        {
    993
    4
            Ok(ids) => ids,
    994
    0
            Err(e) => {
    995
    0
                state.metrics.record_failure();
    996
    0
                return (StatusCode::BAD_REQUEST, Json(ErrorResponse { error: e })).into_response();
    997
            },
    998
        };
    999
    1000
        // Handle streaming for registry models
    1001
    4
        if request.stream {
    1002
    0
            let generated_ids: Vec<u32> = token_ids[prompt.len()..].to_vec();
    1003
    0
            let model_name = request.model.clone();
    1004
    0
            let request_id_clone = request_id.clone();
    1005
    0
            let completion_tokens = generated_ids.len();
    1006
    1007
    0
            let stream = async_stream::stream! {
    1008
                // Send initial chunk with role
    1009
                let initial = ChatCompletionChunk::initial(&request_id_clone, &model_name);
    1010
                if let Ok(data) = serde_json::to_string(&initial) {
    1011
                    yield Ok::<_, Infallible>(Event::default().data(data));
    1012
                }
    1013
    1014
                // Stream tokens one by one
    1015
                for &token_id in &generated_ids {
    1016
                    if let Ok(text) = tokenizer.decode(&[token_id]) {
    1017
                        if !text.is_empty() {
    1018
                            let chunk = ChatCompletionChunk::content(&request_id_clone, &model_name, &text);
    1019
                            if let Ok(data) = serde_json::to_string(&chunk) {
    1020
                                yield Ok(Event::default().data(data));
    1021
                            }
    1022
                        }
    1023
                    }
    1024
                }
    1025
    1026
                // Send final chunk with finish reason
    1027
                let done = ChatCompletionChunk::done(&request_id_clone, &model_name);
    1028
                if let Ok(data) = serde_json::to_string(&done) {
    1029
                    yield Ok(Event::default().data(data));
    1030
                }
    1031
    1032
                // Send [DONE] marker
    1033
                yield Ok(Event::default().data("[DONE]".to_string()));
    1034
            };
    1035
    1036
    0
            state
    1037
    0
                .metrics
    1038
    0
                .record_success(completion_tokens, start.elapsed());
    1039
    0
            return Sse::new(stream).into_response();
    1040
    4
        }
    1041
    1042
        // Non-streaming response
    1043
    4
        let generated_ids = &token_ids[prompt.len()..];
    1044
    4
        let response_text = match tokenizer.decode(generated_ids) {
    1045
    4
            Ok(t) => t,
    1046
    0
            Err(e) => {
    1047
    0
                state.metrics.record_failure();
    1048
    0
                return (
    1049
    0
                    StatusCode::INTERNAL_SERVER_ERROR,
    1050
    0
                    Json(ErrorResponse {
    1051
    0
                        error: e.to_string(),
    1052
    0
                    }),
    1053
    0
                )
    1054
    0
                    .into_response();
    1055
            },
    1056
        };
    1057
    1058
    4
        let completion_tokens = generated_ids.len();
    1059
    4
        let duration = start.elapsed();
    1060
    1061
        // Record successful generation
    1062
    4
        state.metrics.record_success(completion_tokens, duration);
    1063
    1064
        Json(ChatCompletionResponse {
    1065
    4
            id: request_id,
    1066
    4
            object: "chat.completion".to_string(),
    1067
    4
            created: std::time::SystemTime::now()
    1068
    4
                .duration_since(std::time::UNIX_EPOCH)
    1069
    4
                .map(|d| d.as_secs() as i64)
    1070
    4
                .unwrap_or(0),
    1071
    4
            model: request.model.clone(),
    1072
    4
            choices: vec![ChatChoice {
    1073
    4
                index: 0,
    1074
    4
                message: ChatMessage {
    1075
    4
                    role: "assistant".to_string(),
    1076
    4
                    content: response_text,
    1077
    4
                    name: None,
    1078
    4
                },
    1079
    4
                finish_reason: "stop".to_string(),
    1080
    4
            }],
    1081
    4
            usage: Usage {
    1082
    4
                prompt_tokens,
    1083
    4
                completion_tokens,
    1084
    4
                total_tokens: prompt_tokens + completion_tokens,
    1085
    4
            },
    1086
            // Registry models don't support tracing
    1087
    4
            brick_trace: None,
    1088
    4
            step_trace: None,
    1089
    4
            layer_trace: None,
    1090
        })
    1091
    4
        .into_response()
    1092
    6
    }
    1093
    1094
    /// OpenAI-compatible /v1/chat/completions streaming endpoint (SSE)
    1095
    4
    pub async fn openai_chat_completions_stream_handler(
    1096
    4
        State(state): State<AppState>,
    1097
    4
        Json(request): Json<ChatCompletionRequest>,
    1098
    4
    ) -> Result<Sse<impl Stream<Item = Result<Event, Infallible>>>, (StatusCode, Json<ErrorResponse>)> {
    1099
        // Get model and tokenizer
    1100
    4
        let model_id = if request.model == "default" || request.model1.is_empty1() {
    1101
    3
            None
    1102
        } else {
    1103
    1
            Some(request.model.as_str())
    1104
        };
    1105
    1106
    4
        let (model, tokenizer) = state.get_model(model_id).map_err(|e| {0
    1107
    0
            state.metrics.record_failure();
    1108
    0
            (
    1109
    0
                StatusCode::NOT_FOUND,
    1110
    0
                Json(ErrorResponse {
    1111
    0
                    error: e.to_string(),
    1112
    0
                }),
    1113
    0
            )
    1114
    0
        })?;
    1115
    1116
        // Convert chat messages to prompt using model-specific template
    1117
    4
        let prompt_text = format_chat_messages(&request.messages, Some(&request.model));
    1118
    1119
        // Tokenize prompt
    1120
    4
        let prompt_ids = tokenizer.encode(&prompt_text);
    1121
    4
        if prompt_ids.is_empty() {
    1122
    2
            state.metrics.record_failure();
    1123
    2
            return Err((
    1124
    2
                StatusCode::BAD_REQUEST,
    1125
    2
                Json(ErrorResponse {
    1126
    2
                    error: "Messages cannot be empty".to_string(),
    1127
    2
                }),
    1128
    2
            ));
    1129
    2
        }
    1130
    1131
    2
        let prompt_len = prompt_ids.len();
    1132
    1133
        // Convert to usize for model
    1134
    4
        let prompt2: Vec<usize>2 = prompt_ids.iter()2.map2(|&id| id as usize).collect2();
    1135
    1136
        // Build generation config
    1137
    2
        let max_tokens = request.max_tokens.unwrap_or(256);
    1138
    2
        let temperature = request.temperature.unwrap_or(0.7);
    1139
    1140
    2
        let mut config = GenerationConfig::default()
    1141
    2
            .with_max_tokens(max_tokens)
    1142
    2
            .with_temperature(temperature);
    1143
    1144
    2
        if let Some(top_p0) = request.top_p {
    1145
    0
            config.strategy = SamplingStrategy::TopP { p: top_p };
    1146
    2
        }
    1147
    1148
        // Generate request ID
    1149
    2
        let request_id = format!(
    1150
    2
            "chatcmpl-{}",
    1151
    2
            std::time::SystemTime::now()
    1152
    2
                .duration_since(std::time::UNIX_EPOCH)
    1153
    2
                .map(|d| d.as_nanos())
    1154
    2
                .unwrap_or(0)
    1155
        );
    1156
    1157
        // Generate all tokens
    1158
    2
        let generated = model.generate(&prompt, &config).map_err(|e| {0
    1159
    0
            state.metrics.record_failure();
    1160
    0
            (
    1161
    0
                StatusCode::INTERNAL_SERVER_ERROR,
    1162
    0
                Json(ErrorResponse {
    1163
    0
                    error: e.to_string(),
    1164
    0
                }),
    1165
    0
            )
    1166
    0
        })?;
    1167
    1168
        // Convert to u32 for tokenizer
    1169
    2
        let token_ids: Vec<u32> = generated
    1170
    2
            .iter()
    1171
    263
            .filter_map2(|&id| u32::try_from(id).ok())
    1172
    2
            .collect();
    1173
    1174
        // Get only the generated tokens (skip prompt)
    1175
    2
        let generated_ids = token_ids[prompt_len..].to_vec();
    1176
    1177
        // Clone values for move into stream
    1178
    2
        let model_name = request.model.clone();
    1179
    2
        let request_id_clone = request_id.clone();
    1180
    2
        let tokenizer_clone = tokenizer;
    1181
    1182
        // Create SSE stream
    1183
    2
        let stream = async_stream::stream! {
    1184
            // Send initial chunk with role
    1185
            let initial = ChatCompletionChunk::initial(&request_id_clone, &model_name);
    1186
            let data = serde_json::to_string(&initial).unwrap_or_default();
    1187
            yield Ok(Event::default().data(format!("data: {}\n", data)));
    1188
    1189
            // Stream tokens one by one
    1190
            for &token_id in &generated_ids {
    1191
                // Decode single token
    1192
                let text = match tokenizer_clone.decode(&[token_id]) {
    1193
                    Ok(t) => t,
    1194
                    Err(_) => continue,
    1195
                };
    1196
    1197
                let chunk = ChatCompletionChunk::content(&request_id_clone, &model_name, &text);
    1198
                let data = serde_json::to_string(&chunk).unwrap_or_default();
    1199
                yield Ok(Event::default().data(format!("data: {}\n", data)));
    1200
            }
    1201
    1202
            // Send final chunk
    1203
            let done = ChatCompletionChunk::done(&request_id_clone, &model_name);
    1204
            let data = serde_json::to_string(&done).unwrap_or_default();
    1205
            yield Ok(Event::default().data(format!("data: {}\n", data)));
    1206
    1207
            // Send [DONE] marker
    1208
            yield Ok(Event::default().data("data: [DONE]\n".to_string()));
    1209
        };
    1210
    1211
    2
        Ok(Sse::new(stream))
    1212
    4
    }
    1213