openai.rs 45.1 KB
Newer Older
1
2
3
// SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

4
5
6
7
8
9
use std::{
    collections::HashSet,
    sync::Arc,
    time::{SystemTime, UNIX_EPOCH},
};

10
11
use axum::{
    extract::State,
Ryan Olson's avatar
Ryan Olson committed
12
    http::{HeaderMap, StatusCode},
13
14
15
16
17
18
19
    response::{
        sse::{Event, KeepAlive, Sse},
        IntoResponse, Response,
    },
    routing::{get, post},
    Json, Router,
};
Ryan Olson's avatar
Ryan Olson committed
20
21
22
23
24
use dynamo_runtime::{
    pipeline::{AsyncEngineContextProvider, Context},
    protocols::annotated::AnnotationsProvider,
};
use futures::{stream, StreamExt};
25
26
27
use serde::{Deserialize, Serialize};

use super::{
Ryan Olson's avatar
Ryan Olson committed
28
    disconnect::{create_connection_monitor, monitor_for_disconnects, ConnectionHandle},
29
    error::HttpError,
Ryan Olson's avatar
Ryan Olson committed
30
    metrics::{Endpoint, ResponseMetricCollector},
31
    service_v2, RouteDoc,
32
};
33
use crate::preprocessor::LLMMetricAnnotation;
34
use crate::protocols::openai::{
35
36
37
38
    chat_completions::{NvCreateChatCompletionRequest, NvCreateChatCompletionResponse},
    completions::{NvCreateCompletionRequest, NvCreateCompletionResponse},
    embeddings::{NvCreateEmbeddingRequest, NvCreateEmbeddingResponse},
    responses::{NvCreateResponse, NvResponse},
39
};
40
use crate::request_template::RequestTemplate;
41
use crate::types::Annotated;
42
43
use dynamo_runtime::logging::get_distributed_tracing_context;
use tracing::Instrument;
44

Ryan Olson's avatar
Ryan Olson committed
45
46
47
48
49
50
51
pub const DYNAMO_REQUEST_ID_HEADER: &str = "x-dynamo-request-id";

/// Dynamo Annotation for the request ID
pub const ANNOTATION_REQUEST_ID: &str = "request_id";

pub type ErrorResponse = (StatusCode, Json<ErrorMessage>);

52
#[derive(Serialize, Deserialize)]
Ryan Olson's avatar
Ryan Olson committed
53
pub(crate) struct ErrorMessage {
54
55
56
    error: String,
}

Ryan Olson's avatar
Ryan Olson committed
57
impl ErrorMessage {
58
    /// Not Found Error
Ryan Olson's avatar
Ryan Olson committed
59
    pub fn model_not_found() -> ErrorResponse {
60
61
        (
            StatusCode::NOT_FOUND,
Ryan Olson's avatar
Ryan Olson committed
62
            Json(ErrorMessage {
63
64
65
66
67
68
69
                error: "Model not found".to_string(),
            }),
        )
    }

    /// Service Unavailable
    /// This is returned when the service is live, but not ready.
Ryan Olson's avatar
Ryan Olson committed
70
    pub fn _service_unavailable() -> ErrorResponse {
71
72
        (
            StatusCode::SERVICE_UNAVAILABLE,
Ryan Olson's avatar
Ryan Olson committed
73
            Json(ErrorMessage {
74
75
76
77
78
79
80
81
82
                error: "Service is not ready".to_string(),
            }),
        )
    }

    /// Internal Service Error
    /// Return this error when the service encounters an internal error.
    /// We should return a generic message to the client instead of the real error.
    /// Internal Services errors are the result of misconfiguration or bugs in the service.
Ryan Olson's avatar
Ryan Olson committed
83
    pub fn internal_server_error(msg: &str) -> ErrorResponse {
84
85
86
        tracing::error!("Internal server error: {msg}");
        (
            StatusCode::INTERNAL_SERVER_ERROR,
Ryan Olson's avatar
Ryan Olson committed
87
            Json(ErrorMessage {
88
89
90
91
92
                error: msg.to_string(),
            }),
        )
    }

93
94
95
    /// Not Implemented Error
    /// Return this error when the client requests a feature that is not yet implemented.
    /// This should be used for features that are planned but not available.
Ryan Olson's avatar
Ryan Olson committed
96
    pub fn not_implemented_error(msg: &str) -> ErrorResponse {
97
98
99
        tracing::error!("Not Implemented error: {msg}");
        (
            StatusCode::NOT_IMPLEMENTED,
Ryan Olson's avatar
Ryan Olson committed
100
            Json(ErrorMessage {
101
102
103
104
105
                error: msg.to_string(),
            }),
        )
    }

Neelay Shah's avatar
Neelay Shah committed
106
    /// The OAI endpoints call an [`dynamo.runtime::engine::AsyncEngine`] which are specialized to return
107
    /// an [`anyhow::Error`]. This method will convert the [`anyhow::Error`] into an [`HttpError`].
Ryan Olson's avatar
Ryan Olson committed
108
    /// If successful, it will return the [`HttpError`] as an [`ErrorMessage::internal_server_error`]
109
    /// with the details of the error.
Ryan Olson's avatar
Ryan Olson committed
110
    pub fn from_anyhow(err: anyhow::Error, alt_msg: &str) -> ErrorResponse {
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
        // First check for PipelineError::ServiceOverloaded
        if let Some(pipeline_err) =
            err.downcast_ref::<dynamo_runtime::pipeline::error::PipelineError>()
        {
            if matches!(
                pipeline_err,
                dynamo_runtime::pipeline::error::PipelineError::ServiceOverloaded(_)
            ) {
                return (
                    StatusCode::SERVICE_UNAVAILABLE,
                    Json(ErrorMessage {
                        error: pipeline_err.to_string(),
                    }),
                );
            }
        }

        // Then check for HttpError
129
        match err.downcast::<HttpError>() {
Ryan Olson's avatar
Ryan Olson committed
130
131
            Ok(http_error) => ErrorMessage::from_http_error(http_error),
            Err(err) => ErrorMessage::internal_server_error(&format!("{alt_msg}: {err}")),
132
133
134
135
        }
    }

    /// Implementers should only be able to throw 400-499 errors.
Ryan Olson's avatar
Ryan Olson committed
136
    pub fn from_http_error(err: HttpError) -> ErrorResponse {
137
        if err.code < 400 || err.code >= 500 {
Ryan Olson's avatar
Ryan Olson committed
138
            return ErrorMessage::internal_server_error(&err.message);
139
140
        }
        match StatusCode::from_u16(err.code) {
Ryan Olson's avatar
Ryan Olson committed
141
142
            Ok(code) => (code, Json(ErrorMessage { error: err.message })),
            Err(_) => ErrorMessage::internal_server_error(&err.message),
143
144
145
146
        }
    }
}

Ryan Olson's avatar
Ryan Olson committed
147
impl From<HttpError> for ErrorMessage {
148
    fn from(err: HttpError) -> Self {
Ryan Olson's avatar
Ryan Olson committed
149
        ErrorMessage { error: err.message }
150
151
152
    }
}

Ryan Olson's avatar
Ryan Olson committed
153
154
/// Get the request ID from a primary source, or next from the headers, or lastly create a new one if not present
fn get_or_create_request_id(primary: Option<&str>, headers: &HeaderMap) -> String {
155
156
157
158
159
160
161
    // Try to get request id from trace context
    if let Some(trace_context) = get_distributed_tracing_context() {
        if let Some(x_dynamo_request_id) = trace_context.x_dynamo_request_id {
            return x_dynamo_request_id;
        }
    }

Ryan Olson's avatar
Ryan Olson committed
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
    // Try to get the request ID from the primary source
    if let Some(primary) = primary {
        if let Ok(uuid) = uuid::Uuid::parse_str(primary) {
            return uuid.to_string();
        }
    }

    // Try to get the request ID header as a string slice
    let request_id_opt = headers
        .get(DYNAMO_REQUEST_ID_HEADER)
        .and_then(|h| h.to_str().ok());

    // Try to parse the request ID as a UUID, or generate a new one if missing/invalid
    let uuid = match request_id_opt {
        Some(request_id) => {
            uuid::Uuid::parse_str(request_id).unwrap_or_else(|_| uuid::Uuid::new_v4())
        }
        None => uuid::Uuid::new_v4(),
    };

    uuid.to_string()
}

185
186
187
188
189
190
191
192
/// OpenAI Completions Request Handler
///
/// This method will handle the incoming request for the `/v1/completions endpoint`. The endpoint is a "source"
/// for an [`super::OpenAICompletionsStreamingEngine`] and will return a stream of
/// responses which will be forward to the client.
///
/// Note: For all requests, streaming or non-streaming, we always call the engine with streaming enabled. For
/// non-streaming requests, we will fold the stream into a single response as part of this handler.
Ryan Olson's avatar
Ryan Olson committed
193
async fn handler_completions(
194
    State(state): State<Arc<service_v2::State>>,
Ryan Olson's avatar
Ryan Olson committed
195
    headers: HeaderMap,
196
    Json(request): Json<NvCreateCompletionRequest>,
Ryan Olson's avatar
Ryan Olson committed
197
198
199
200
201
202
203
204
205
206
207
208
209
210
) -> Result<Response, ErrorResponse> {
    // return a 503 if the service is not ready
    check_ready(&state)?;

    // create the context for the request
    let request_id = get_or_create_request_id(request.inner.user.as_deref(), &headers);
    let request = Context::with_id(request, request_id);
    let context = request.context();

    // create the connection handles
    let (mut connection_handle, stream_handle) = create_connection_monitor(context.clone()).await;

    // possibly long running task
    // if this returns a streaming response, the stream handle will be armed and captured by the response stream
211
    let response = tokio::spawn(completions(state, request, stream_handle).in_current_span())
Ryan Olson's avatar
Ryan Olson committed
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
        .await
        .map_err(|e| {
            ErrorMessage::internal_server_error(&format!(
                "Failed to await chat completions task: {:?}",
                e,
            ))
        })?;

    // if we got here, then we will return a response and the potentially long running task has completed successfully
    // without need to be cancelled.
    connection_handle.disarm();

    response
}

#[tracing::instrument(skip_all)]
async fn completions(
    state: Arc<service_v2::State>,
    request: Context<NvCreateCompletionRequest>,
    stream_handle: ConnectionHandle,
) -> Result<Response, ErrorResponse> {
233
234
235
236
237
238
239
    // return a 503 if the service is not ready
    check_ready(&state)?;

    // todo - extract distributed tracing id and context id from headers
    let request_id = uuid::Uuid::new_v4().to_string();

    // todo - decide on default
240
    let streaming = request.inner.stream.unwrap_or(false);
241
242

    // update the request to always stream
Ryan Olson's avatar
Ryan Olson committed
243
244
245
246
    let request = request.map(|mut req| {
        req.inner.stream = Some(true);
        req
    });
247

248
249
    // todo - make the protocols be optional for model name
    // todo - when optional, if none, apply a default
250
    let model = &request.inner.model;
251
252
253

    // todo - error handling should be more robust
    let engine = state
254
        .manager()
255
        .get_completions_engine(model)
Ryan Olson's avatar
Ryan Olson committed
256
        .map_err(|_| ErrorMessage::model_not_found())?;
257

258
    let mut inflight_guard =
259
260
261
        state
            .metrics_clone()
            .create_inflight_guard(model, Endpoint::Completions, streaming);
262

263
264
    let mut response_collector = state.metrics_clone().create_response_collector(model);

Ryan Olson's avatar
Ryan Olson committed
265
266
    // prepare to process any annotations
    let annotations = request.annotations();
267
268
269
270
271

    // issue the generate call on the engine
    let stream = engine
        .generate(request)
        .await
Ryan Olson's avatar
Ryan Olson committed
272
        .map_err(|e| ErrorMessage::from_anyhow(e, "Failed to generate completions"))?;
273
274
275
276

    // capture the context to cancel the stream if the client disconnects
    let ctx = stream.context();

Ryan Olson's avatar
Ryan Olson committed
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
    let annotations = annotations.map_or(Vec::new(), |annotations| {
        annotations
            .iter()
            .filter_map(|annotation| {
                if annotation == ANNOTATION_REQUEST_ID {
                    Annotated::<NvCreateCompletionResponse>::from_annotation(
                        ANNOTATION_REQUEST_ID,
                        &request_id,
                    )
                    .ok()
                } else {
                    None
                }
            })
            .collect::<Vec<_>>()
    });

    // apply any annotations to the front of the stream
    let stream = stream::iter(annotations).chain(stream);
296
297

    if streaming {
298
299
300
        let stream = stream.map(move |response| {
            process_event_converter(EventConverter::from(response), &mut response_collector)
        });
Ryan Olson's avatar
Ryan Olson committed
301
        let stream = monitor_for_disconnects(stream, ctx, inflight_guard, stream_handle);
302

303
304
        let mut sse_stream = Sse::new(stream);

305
        if let Some(keep_alive) = state.sse_keep_alive() {
306
307
308
309
            sse_stream = sse_stream.keep_alive(KeepAlive::default().interval(keep_alive));
        }

        Ok(sse_stream.into_response())
310
    } else {
311
312
313
314
315
        // Tap the stream to collect metrics for non-streaming requests without altering items
        let stream = stream.inspect(move |response| {
            process_metrics_only(response, &mut response_collector);
        });

Ryan Olson's avatar
Ryan Olson committed
316
        let response = NvCreateCompletionResponse::from_annotated_stream(stream)
317
318
319
320
321
322
323
            .await
            .map_err(|e| {
                tracing::error!(
                    "Failed to fold completions stream for {}: {:?}",
                    request_id,
                    e
                );
Ryan Olson's avatar
Ryan Olson committed
324
                ErrorMessage::internal_server_error("Failed to fold completions stream")
325
326
            })?;

327
        inflight_guard.mark_ok();
328
329
330
331
        Ok(Json(response).into_response())
    }
}

332
333
#[tracing::instrument(skip_all)]
async fn embeddings(
334
335
    State(state): State<Arc<service_v2::State>>,
    Json(request): Json<NvCreateEmbeddingRequest>,
Ryan Olson's avatar
Ryan Olson committed
336
) -> Result<Response, ErrorResponse> {
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
    // return a 503 if the service is not ready
    check_ready(&state)?;

    // todo - extract distributed tracing id and context id from headers
    let request_id = uuid::Uuid::new_v4().to_string();

    // Embeddings are typically not streamed, so we default to non-streaming
    let streaming = false;

    // todo - make the protocols be optional for model name
    // todo - when optional, if none, apply a default
    let model = &request.inner.model;

    // todo - error handling should be more robust
    let engine = state
        .manager()
        .get_embeddings_engine(model)
Ryan Olson's avatar
Ryan Olson committed
354
        .map_err(|_| ErrorMessage::model_not_found())?;
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369

    // this will increment the inflight gauge for the model
    let mut inflight =
        state
            .metrics_clone()
            .create_inflight_guard(model, Endpoint::Embeddings, streaming);

    // setup context
    // todo - inherit request_id from distributed trace details
    let request = Context::with_id(request, request_id.clone());

    // issue the generate call on the engine
    let stream = engine
        .generate(request)
        .await
Ryan Olson's avatar
Ryan Olson committed
370
        .map_err(|e| ErrorMessage::from_anyhow(e, "Failed to generate embeddings"))?;
371
372
373

    // Embeddings are typically returned as a single response (non-streaming)
    // so we fold the stream into a single response
Ryan Olson's avatar
Ryan Olson committed
374
    let response = NvCreateEmbeddingResponse::from_annotated_stream(stream)
375
376
377
378
379
380
381
        .await
        .map_err(|e| {
            tracing::error!(
                "Failed to fold embeddings stream for {}: {:?}",
                request_id,
                e
            );
Ryan Olson's avatar
Ryan Olson committed
382
            ErrorMessage::internal_server_error("Failed to fold embeddings stream")
383
384
385
386
        })?;

    inflight.mark_ok();
    Ok(Json(response).into_response())
387
388
}

Ryan Olson's avatar
Ryan Olson committed
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
async fn handler_chat_completions(
    State((state, template)): State<(Arc<service_v2::State>, Option<RequestTemplate>)>,
    headers: HeaderMap,
    Json(request): Json<NvCreateChatCompletionRequest>,
) -> Result<Response, ErrorResponse> {
    // return a 503 if the service is not ready
    check_ready(&state)?;

    // create the context for the request
    let request_id = get_or_create_request_id(request.inner.user.as_deref(), &headers);
    let request = Context::with_id(request, request_id);
    let context = request.context();

    // create the connection handles
    let (mut connection_handle, stream_handle) = create_connection_monitor(context.clone()).await;

405
406
407
408
409
410
411
412
413
    let response =
        tokio::spawn(chat_completions(state, template, request, stream_handle).in_current_span())
            .await
            .map_err(|e| {
                ErrorMessage::internal_server_error(&format!(
                    "Failed to await chat completions task: {:?}",
                    e,
                ))
            })?;
Ryan Olson's avatar
Ryan Olson committed
414
415
416
417
418
419
420
421

    // if we got here, then we will return a response and the potentially long running task has completed successfully
    // without need to be cancelled.
    connection_handle.disarm();

    response
}

422
423
424
425
426
427
428
429
430
/// OpenAI Chat Completions Request Handler
///
/// This method will handle the incoming request for the /v1/chat/completions endpoint. The endpoint is a "source"
/// for an [`super::OpenAIChatCompletionsStreamingEngine`] and will return a stream of responses which will be
/// forward to the client.
///
/// Note: For all requests, streaming or non-streaming, we always call the engine with streaming enabled. For
/// non-streaming requests, we will fold the stream into a single response as part of this handler.
async fn chat_completions(
Ryan Olson's avatar
Ryan Olson committed
431
432
433
434
435
    state: Arc<service_v2::State>,
    template: Option<RequestTemplate>,
    mut request: Context<NvCreateChatCompletionRequest>,
    mut stream_handle: ConnectionHandle,
) -> Result<Response, ErrorResponse> {
436
437
438
    // return a 503 if the service is not ready
    check_ready(&state)?;

Ryan Olson's avatar
Ryan Olson committed
439
440
    let request_id = request.id().to_string();

441
442
443
444
    // Handle unsupported fields - if Some(resp) is returned by
    // validate_chat_completion_unsupported_fields,
    // then a field was used that is unsupported. We will log an error message
    // and early return a 501 NOT_IMPLEMENTED status code. Otherwise, proceeed.
Ryan Olson's avatar
Ryan Olson committed
445
    validate_chat_completion_unsupported_fields(&request)?;
446

447
448
449
    // Handle required fields like messages shouldn't be empty.
    validate_chat_completion_required_fields(&request)?;

450
451
452
453
454
455
456
457
458
459
460
461
    // Apply template values if present
    if let Some(template) = template {
        if request.inner.model.is_empty() {
            request.inner.model = template.model.clone();
        }
        if request.inner.temperature.unwrap_or(0.0) == 0.0 {
            request.inner.temperature = Some(template.temperature);
        }
        if request.inner.max_completion_tokens.unwrap_or(0) == 0 {
            request.inner.max_completion_tokens = Some(template.max_completion_tokens);
        }
    }
Ryan Olson's avatar
Ryan Olson committed
462
    tracing::trace!("Received chat completions request: {:?}", request.content());
463
464

    // todo - decide on default
Paul Hendricks's avatar
Paul Hendricks committed
465
    let streaming = request.inner.stream.unwrap_or(false);
466
467

    // update the request to always stream
Ryan Olson's avatar
Ryan Olson committed
468
469
470
471
    let request = request.map(|mut req| {
        req.inner.stream = Some(true);
        req
    });
472
473
474

    // todo - make the protocols be optional for model name
    // todo - when optional, if none, apply a default
Paul Hendricks's avatar
Paul Hendricks committed
475
    let model = &request.inner.model;
476
477
478
479
480

    // todo - determine the proper error code for when a request model is not present
    tracing::trace!("Getting chat completions engine for model: {}", model);

    let engine = state
481
        .manager()
482
        .get_chat_completions_engine(model)
Ryan Olson's avatar
Ryan Olson committed
483
        .map_err(|_| ErrorMessage::model_not_found())?;
484

485
    let mut inflight_guard =
486
487
488
        state
            .metrics_clone()
            .create_inflight_guard(model, Endpoint::ChatCompletions, streaming);
489

490
491
    let mut response_collector = state.metrics_clone().create_response_collector(model);

492
    tracing::trace!("Issuing generate call for chat completions");
Ryan Olson's avatar
Ryan Olson committed
493
    let annotations = request.annotations();
494
495
496
497
498

    // issue the generate call on the engine
    let stream = engine
        .generate(request)
        .await
Ryan Olson's avatar
Ryan Olson committed
499
        .map_err(|e| ErrorMessage::from_anyhow(e, "Failed to generate completions"))?;
500
501
502
503

    // capture the context to cancel the stream if the client disconnects
    let ctx = stream.context();

Ryan Olson's avatar
Ryan Olson committed
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
    // prepare any requested annotations
    let annotations = annotations.map_or(Vec::new(), |annotations| {
        annotations
            .iter()
            .filter_map(|annotation| {
                if annotation == ANNOTATION_REQUEST_ID {
                    Annotated::from_annotation(ANNOTATION_REQUEST_ID, &request_id).ok()
                } else {
                    None
                }
            })
            .collect::<Vec<_>>()
    });

    // apply any annotations to the front of the stream
    let stream = stream::iter(annotations).chain(stream);

521
522
523
524
    // todo - tap the stream and propagate request level metrics
    // note - we might do this as part of the post processing set to make it more generic

    if streaming {
Ryan Olson's avatar
Ryan Olson committed
525
526
        stream_handle.arm();

527
528
529
        let stream = stream.map(move |response| {
            process_event_converter(EventConverter::from(response), &mut response_collector)
        });
Ryan Olson's avatar
Ryan Olson committed
530
        let stream = monitor_for_disconnects(stream, ctx, inflight_guard, stream_handle);
531

532
533
        let mut sse_stream = Sse::new(stream);

534
        if let Some(keep_alive) = state.sse_keep_alive() {
535
536
537
538
            sse_stream = sse_stream.keep_alive(KeepAlive::default().interval(keep_alive));
        }

        Ok(sse_stream.into_response())
539
    } else {
540
541
542
543
        let stream = stream.inspect(move |response| {
            process_metrics_only(response, &mut response_collector);
        });

Ryan Olson's avatar
Ryan Olson committed
544
        let response = NvCreateChatCompletionResponse::from_annotated_stream(stream)
545
546
547
548
549
550
551
            .await
            .map_err(|e| {
                tracing::error!(
                    request_id,
                    "Failed to fold chat completions stream for: {:?}",
                    e
                );
Ryan Olson's avatar
Ryan Olson committed
552
                ErrorMessage::internal_server_error(&format!(
553
554
555
556
557
                    "Failed to fold chat completions stream: {}",
                    e
                ))
            })?;

558
        inflight_guard.mark_ok();
559
560
561
562
        Ok(Json(response).into_response())
    }
}

563
564
565
566
567
/// Checks for unsupported fields in the request.
/// Returns Some(response) if unsupported fields are present.
#[allow(deprecated)]
pub fn validate_chat_completion_unsupported_fields(
    request: &NvCreateChatCompletionRequest,
Ryan Olson's avatar
Ryan Olson committed
568
) -> Result<(), ErrorResponse> {
569
570
571
    let inner = &request.inner;

    if inner.parallel_tool_calls == Some(true) {
Ryan Olson's avatar
Ryan Olson committed
572
        return Err(ErrorMessage::not_implemented_error(
573
574
575
576
577
            "`parallel_tool_calls: true` is not supported.",
        ));
    }

    if inner.stream == Some(true) && inner.tools.is_some() {
Ryan Olson's avatar
Ryan Olson committed
578
        return Err(ErrorMessage::not_implemented_error(
579
580
581
582
583
            "`stream: true` is not supported when `tools` are provided.",
        ));
    }

    if inner.function_call.is_some() {
Ryan Olson's avatar
Ryan Olson committed
584
        return Err(ErrorMessage::not_implemented_error(
585
586
587
588
589
            "`function_call` is deprecated. Please migrate to use `tool_choice` instead.",
        ));
    }

    if inner.functions.is_some() {
Ryan Olson's avatar
Ryan Olson committed
590
        return Err(ErrorMessage::not_implemented_error(
591
592
593
594
            "`functions` is deprecated. Please migrate to use `tools` instead.",
        ));
    }

Ryan Olson's avatar
Ryan Olson committed
595
    Ok(())
596
597
}

598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
/// Validates that required fields are present and valid in the chat completion request
pub fn validate_chat_completion_required_fields(
    request: &NvCreateChatCompletionRequest,
) -> Result<(), ErrorResponse> {
    let inner = &request.inner;

    if inner.messages.is_empty() {
        return Err(ErrorMessage::from_http_error(HttpError {
            code: 400,
            message: "The 'messages' field cannot be empty. At least one message is required."
                .to_string(),
        }));
    }

    Ok(())
}

615
616
617
/// OpenAI Responses Request Handler
///
/// This method will handle the incoming request for the /v1/responses endpoint.
Ryan Olson's avatar
Ryan Olson committed
618
async fn handler_responses(
619
    State((state, template)): State<(Arc<service_v2::State>, Option<RequestTemplate>)>,
Ryan Olson's avatar
Ryan Olson committed
620
621
622
623
624
625
626
627
628
629
630
631
632
633
    headers: HeaderMap,
    Json(request): Json<NvCreateResponse>,
) -> Result<Response, ErrorResponse> {
    // return a 503 if the service is not ready
    check_ready(&state)?;

    // create the context for the request
    let request_id = get_or_create_request_id(request.inner.user.as_deref(), &headers);
    let request = Context::with_id(request, request_id);
    let context = request.context();

    // create the connection handles
    let (mut connection_handle, _stream_handle) = create_connection_monitor(context.clone()).await;

634
    let response = tokio::spawn(responses(state, template, request).in_current_span())
Ryan Olson's avatar
Ryan Olson committed
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
        .await
        .map_err(|e| {
            ErrorMessage::internal_server_error(&format!(
                "Failed to await chat completions task: {:?}",
                e,
            ))
        })?;

    // if we got here, then we will return a response and the potentially long running task has completed successfully
    // without need to be cancelled.
    connection_handle.disarm();

    response
}

#[tracing::instrument(level = "debug", skip_all, fields(request_id = %request.id()))]
async fn responses(
    state: Arc<service_v2::State>,
    template: Option<RequestTemplate>,
    mut request: Context<NvCreateResponse>,
) -> Result<Response, ErrorResponse> {
656
657
658
659
660
661
    // return a 503 if the service is not ready
    check_ready(&state)?;

    // Handle unsupported fields - if Some(resp) is returned by validate_unsupported_fields,
    // then a field was used that is unsupported. We will log an error message
    // and early return a 501 NOT_IMPLEMENTED status code. Otherwise, proceeed.
662
    if let Some(resp) = validate_response_unsupported_fields(&request) {
663
664
665
666
667
668
669
        return Ok(resp.into_response());
    }

    // Handle non-text (image, audio, file) inputs - if Some(resp) is returned by
    // validate_input_is_text_only, then we are handling something other than Input::Text(_).
    // We will log an error message and early return a 501 NOT_IMPLEMENTED status code.
    // Otherwise, proceeed.
670
    if let Some(resp) = validate_response_input_is_text_only(&request) {
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
        return Ok(resp.into_response());
    }

    // Apply template values if present
    if let Some(template) = template {
        if request.inner.model.is_empty() {
            request.inner.model = template.model.clone();
        }
        if request.inner.temperature.unwrap_or(0.0) == 0.0 {
            request.inner.temperature = Some(template.temperature);
        }
        if request.inner.max_output_tokens.unwrap_or(0) == 0 {
            request.inner.max_output_tokens = Some(template.max_completion_tokens);
        }
    }
    tracing::trace!("Received chat completions request: {:?}", request.inner);

Ryan Olson's avatar
Ryan Olson committed
688
689
    let request_id = request.id().to_string();
    let (request, context) = request.into_parts();
690

Ryan Olson's avatar
Ryan Olson committed
691
    let mut request: NvCreateChatCompletionRequest = request.try_into().map_err(|e| {
692
693
694
695
696
        tracing::error!(
            request_id,
            "Failed to convert NvCreateResponse to NvCreateChatCompletionRequest: {:?}",
            e
        );
Ryan Olson's avatar
Ryan Olson committed
697
        ErrorMessage::not_implemented_error(&format!(
698
699
700
701
702
            "Only Input::Text(_) is currently supported: {}",
            e
        ))
    })?;

Ryan Olson's avatar
Ryan Olson committed
703
704
705
706
707
    let request = context.map(|mut _req| {
        request.inner.stream = Some(false);
        request
    });

708
709
710
711
712
713
714
    let model = &request.inner.model;

    tracing::trace!("Getting chat completions engine for model: {}", model);

    let engine = state
        .manager()
        .get_chat_completions_engine(model)
Ryan Olson's avatar
Ryan Olson committed
715
        .map_err(|_| ErrorMessage::model_not_found())?;
716
717
718
719
720
721
722
723
724
725
726
727
728
729

    let mut inflight_guard =
        state
            .metrics_clone()
            .create_inflight_guard(model, Endpoint::Responses, false);

    let _response_collector = state.metrics_clone().create_response_collector(model);

    tracing::trace!("Issuing generate call for chat completions");

    // issue the generate call on the engine
    let stream = engine
        .generate(request)
        .await
Ryan Olson's avatar
Ryan Olson committed
730
        .map_err(|e| ErrorMessage::from_anyhow(e, "Failed to generate completions"))?;
731
732

    // TODO: handle streaming, currently just unary
Ryan Olson's avatar
Ryan Olson committed
733
    let response = NvCreateChatCompletionResponse::from_annotated_stream(stream)
734
735
736
737
738
739
740
        .await
        .map_err(|e| {
            tracing::error!(
                request_id,
                "Failed to fold chat completions stream for: {:?}",
                e
            );
Ryan Olson's avatar
Ryan Olson committed
741
            ErrorMessage::internal_server_error(&format!(
742
743
744
745
746
747
748
749
750
751
752
753
                "Failed to fold chat completions stream: {}",
                e
            ))
        })?;

    // Convert NvCreateChatCompletionResponse --> NvResponse
    let response: NvResponse = response.try_into().map_err(|e| {
        tracing::error!(
            request_id,
            "Failed to convert NvCreateChatCompletionResponse to NvResponse: {:?}",
            e
        );
Ryan Olson's avatar
Ryan Olson committed
754
        ErrorMessage::internal_server_error("Failed to convert internal response")
755
756
757
758
759
760
761
    })?;

    inflight_guard.mark_ok();

    Ok(Json(response).into_response())
}

762
763
764
pub fn validate_response_input_is_text_only(
    request: &NvCreateResponse,
) -> Option<impl IntoResponse> {
765
    match &request.inner.input {
766
        dynamo_async_openai::types::responses::Input::Text(_) => None,
Ryan Olson's avatar
Ryan Olson committed
767
        _ => Some(ErrorMessage::not_implemented_error("Only `Input::Text` is supported. Structured, multimedia, or custom input types are not yet implemented.")),
768
769
770
771
772
    }
}

/// Checks for unsupported fields in the request.
/// Returns Some(response) if unsupported fields are present.
773
774
775
pub fn validate_response_unsupported_fields(
    request: &NvCreateResponse,
) -> Option<impl IntoResponse> {
776
777
778
    let inner = &request.inner;

    if inner.background == Some(true) {
Ryan Olson's avatar
Ryan Olson committed
779
        return Some(ErrorMessage::not_implemented_error(
780
781
782
783
            "`background: true` is not supported.",
        ));
    }
    if inner.include.is_some() {
Ryan Olson's avatar
Ryan Olson committed
784
        return Some(ErrorMessage::not_implemented_error(
785
786
787
788
            "`include` is not supported.",
        ));
    }
    if inner.instructions.is_some() {
Ryan Olson's avatar
Ryan Olson committed
789
        return Some(ErrorMessage::not_implemented_error(
790
791
792
793
            "`instructions` is not supported.",
        ));
    }
    if inner.max_tool_calls.is_some() {
Ryan Olson's avatar
Ryan Olson committed
794
        return Some(ErrorMessage::not_implemented_error(
795
796
797
798
            "`max_tool_calls` is not supported.",
        ));
    }
    if inner.metadata.is_some() {
Ryan Olson's avatar
Ryan Olson committed
799
        return Some(ErrorMessage::not_implemented_error(
800
801
802
803
            "`metadata` is not supported.",
        ));
    }
    if inner.parallel_tool_calls == Some(true) {
Ryan Olson's avatar
Ryan Olson committed
804
        return Some(ErrorMessage::not_implemented_error(
805
806
807
808
            "`parallel_tool_calls: true` is not supported.",
        ));
    }
    if inner.previous_response_id.is_some() {
Ryan Olson's avatar
Ryan Olson committed
809
        return Some(ErrorMessage::not_implemented_error(
810
811
812
813
            "`previous_response_id` is not supported.",
        ));
    }
    if inner.prompt.is_some() {
Ryan Olson's avatar
Ryan Olson committed
814
        return Some(ErrorMessage::not_implemented_error(
815
816
817
818
            "`prompt` is not supported.",
        ));
    }
    if inner.reasoning.is_some() {
Ryan Olson's avatar
Ryan Olson committed
819
        return Some(ErrorMessage::not_implemented_error(
820
821
822
823
            "`reasoning` is not supported.",
        ));
    }
    if inner.service_tier.is_some() {
Ryan Olson's avatar
Ryan Olson committed
824
        return Some(ErrorMessage::not_implemented_error(
825
826
827
828
            "`service_tier` is not supported.",
        ));
    }
    if inner.store == Some(true) {
Ryan Olson's avatar
Ryan Olson committed
829
        return Some(ErrorMessage::not_implemented_error(
830
831
832
833
            "`store: true` is not supported.",
        ));
    }
    if inner.stream == Some(true) {
Ryan Olson's avatar
Ryan Olson committed
834
        return Some(ErrorMessage::not_implemented_error(
835
836
837
838
            "`stream: true` is not supported.",
        ));
    }
    if inner.text.is_some() {
Ryan Olson's avatar
Ryan Olson committed
839
        return Some(ErrorMessage::not_implemented_error(
840
841
842
843
            "`text` is not supported.",
        ));
    }
    if inner.tool_choice.is_some() {
Ryan Olson's avatar
Ryan Olson committed
844
        return Some(ErrorMessage::not_implemented_error(
845
846
847
848
            "`tool_choice` is not supported.",
        ));
    }
    if inner.tools.is_some() {
Ryan Olson's avatar
Ryan Olson committed
849
        return Some(ErrorMessage::not_implemented_error(
850
851
852
853
            "`tools` is not supported.",
        ));
    }
    if inner.truncation.is_some() {
Ryan Olson's avatar
Ryan Olson committed
854
        return Some(ErrorMessage::not_implemented_error(
855
856
857
858
            "`truncation` is not supported.",
        ));
    }
    if inner.user.is_some() {
Ryan Olson's avatar
Ryan Olson committed
859
        return Some(ErrorMessage::not_implemented_error(
860
861
862
863
864
865
866
            "`user` is not supported.",
        ));
    }

    None
}

867
868
// todo - abstract this to the top level lib.rs to be reused
// todo - move the service_observer to its own state/arc
Ryan Olson's avatar
Ryan Olson committed
869
fn check_ready(_state: &Arc<service_v2::State>) -> Result<(), ErrorResponse> {
870
    // if state.service_observer.stage() != ServiceStage::Ready {
Ryan Olson's avatar
Ryan Olson committed
871
    //     return Err(ErrorMessage::service_unavailable());
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
    // }
    Ok(())
}

/// openai compatible format
/// Example:
/// {
///  "object": "list",
///  "data": [
///    {
///      "id": "model-id-0",
///      "object": "model",
///      "created": 1686935002,
///      "owned_by": "organization-owner"
///    },
///    ]
/// }
async fn list_models_openai(
890
    State(state): State<Arc<service_v2::State>>,
Ryan Olson's avatar
Ryan Olson committed
891
) -> Result<Response, ErrorResponse> {
892
893
894
895
896
897
898
899
    check_ready(&state)?;

    let created = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_secs();
    let mut data = Vec::new();

900
901
    let models: HashSet<String> = state.manager().model_display_names();
    for model_name in models {
902
        data.push(ModelListing {
903
            id: model_name.clone(),
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
            object: "object",
            created,                        // Where would this come from? The GGUF?
            owned_by: "nvidia".to_string(), // Get organization from GGUF
        });
    }

    let out = ListModelOpenAI {
        object: "list",
        data,
    };
    Ok(Json(out).into_response())
}

#[derive(Serialize)]
struct ListModelOpenAI {
    object: &'static str, // always "list"
    data: Vec<ModelListing>,
}

#[derive(Serialize)]
struct ModelListing {
    id: String,
    object: &'static str, // always "object"
    created: u64,         //  Seconds since epoch
    owned_by: String,
}

struct EventConverter<T>(Annotated<T>);

impl<T> From<Annotated<T>> for EventConverter<T> {
    fn from(annotated: Annotated<T>) -> Self {
        EventConverter(annotated)
    }
}

939
940
941
942
943
944
945
946
947
948
949
fn process_metrics_only<T>(
    annotated: &Annotated<T>,
    response_collector: &mut ResponseMetricCollector,
) {
    // update metrics
    if let Ok(Some(metrics)) = LLMMetricAnnotation::from_annotation(annotated) {
        response_collector.observe_current_osl(metrics.output_tokens);
        response_collector.observe_response(metrics.input_tokens, metrics.chunk_tokens);
    }
}

950
951
952
953
fn process_event_converter<T: Serialize>(
    annotated: EventConverter<T>,
    response_collector: &mut ResponseMetricCollector,
) -> Result<Event, axum::Error> {
954
    let mut annotated = annotated.0;
955

956
957
958
959
    // update metrics
    if let Ok(Some(metrics)) = LLMMetricAnnotation::from_annotation(&annotated) {
        response_collector.observe_current_osl(metrics.output_tokens);
        response_collector.observe_response(metrics.input_tokens, metrics.chunk_tokens);
960
961
962
963
964
965
966

        // Chomp the LLMMetricAnnotation so it's not returned in the response stream
        // TODO: add a flag to control what is returned in the SSE stream
        if annotated.event.as_deref() == Some(crate::preprocessor::ANNOTATION_LLM_METRICS) {
            annotated.event = None;
            annotated.comment = None;
        }
967
968
    }

969
    let mut event = Event::default();
970

971
972
973
    if let Some(data) = annotated.data {
        event = event.json_data(data)?;
    }
974

975
976
977
978
979
980
    if let Some(msg) = annotated.event {
        if msg == "error" {
            let msgs = annotated
                .comment
                .unwrap_or_else(|| vec!["unspecified error".to_string()]);
            return Err(axum::Error::new(msgs.join(" -- ")));
981
        }
982
983
        event = event.event(msg);
    }
984

985
986
987
988
    if let Some(comments) = annotated.comment {
        for comment in comments {
            event = event.comment(comment);
        }
989
    }
990
991

    Ok(event)
992
993
994
995
996
}

/// Create an Axum [`Router`] for the OpenAI API Completions endpoint
/// If not path is provided, the default path is `/v1/completions`
pub fn completions_router(
997
    state: Arc<service_v2::State>,
998
999
1000
1001
1002
    path: Option<String>,
) -> (Vec<RouteDoc>, Router) {
    let path = path.unwrap_or("/v1/completions".to_string());
    let doc = RouteDoc::new(axum::http::Method::POST, &path);
    let router = Router::new()
Ryan Olson's avatar
Ryan Olson committed
1003
        .route(&path, post(handler_completions))
1004
1005
1006
1007
1008
1009
1010
        .with_state(state);
    (vec![doc], router)
}

/// Create an Axum [`Router`] for the OpenAI API Chat Completions endpoint
/// If not path is provided, the default path is `/v1/chat/completions`
pub fn chat_completions_router(
1011
    state: Arc<service_v2::State>,
1012
    template: Option<RequestTemplate>,
1013
1014
1015
1016
1017
    path: Option<String>,
) -> (Vec<RouteDoc>, Router) {
    let path = path.unwrap_or("/v1/chat/completions".to_string());
    let doc = RouteDoc::new(axum::http::Method::POST, &path);
    let router = Router::new()
Ryan Olson's avatar
Ryan Olson committed
1018
        .route(&path, post(handler_chat_completions))
1019
        .with_state((state, template));
1020
1021
1022
    (vec![doc], router)
}

1023
1024
1025
/// Create an Axum [`Router`] for the OpenAI API Embeddings endpoint
/// If not path is provided, the default path is `/v1/embeddings`
pub fn embeddings_router(
1026
    state: Arc<service_v2::State>,
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
    path: Option<String>,
) -> (Vec<RouteDoc>, Router) {
    let path = path.unwrap_or("/v1/embeddings".to_string());
    let doc = RouteDoc::new(axum::http::Method::POST, &path);
    let router = Router::new()
        .route(&path, post(embeddings))
        .with_state(state);
    (vec![doc], router)
}

1037
1038
/// List Models
pub fn list_models_router(
1039
    state: Arc<service_v2::State>,
1040
1041
1042
    path: Option<String>,
) -> (Vec<RouteDoc>, Router) {
    // Standard OpenAI compatible list models endpoint
1043
    let openai_path = path.unwrap_or("/v1/models".to_string());
1044
1045
1046
1047
1048
1049
    let doc_for_openai = RouteDoc::new(axum::http::Method::GET, &openai_path);

    let router = Router::new()
        .route(&openai_path, get(list_models_openai))
        .with_state(state);

1050
    (vec![doc_for_openai], router)
1051
1052
}

1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
/// Create an Axum [`Router`] for the OpenAI API Responses endpoint
/// If not path is provided, the default path is `/v1/responses`
pub fn responses_router(
    state: Arc<service_v2::State>,
    template: Option<RequestTemplate>,
    path: Option<String>,
) -> (Vec<RouteDoc>, Router) {
    let path = path.unwrap_or("/v1/responses".to_string());
    let doc = RouteDoc::new(axum::http::Method::POST, &path);
    let router = Router::new()
Ryan Olson's avatar
Ryan Olson committed
1063
        .route(&path, post(handler_responses))
1064
1065
1066
1067
        .with_state((state, template));
    (vec![doc], router)
}

1068
1069
#[cfg(test)]
mod tests {
1070
1071
    use std::collections::HashMap;

1072
    use dynamo_async_openai::types::responses::{
1073
1074
1075
1076
        CreateResponse, Input, InputContent, InputItem, InputMessage, PromptConfig,
        Role as ResponseRole, ServiceTier, TextConfig, TextResponseFormat, ToolChoice,
        ToolChoiceMode, Truncation,
    };
1077
    use dynamo_async_openai::types::{
1078
1079
1080
        ChatCompletionRequestMessage, ChatCompletionRequestUserMessage,
        ChatCompletionRequestUserMessageContent, CreateChatCompletionRequest,
    };
1081
1082

    use super::*;
1083
1084
    use crate::discovery::ModelManagerError;
    use crate::protocols::openai::responses::NvCreateResponse;
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095

    const BACKUP_ERROR_MESSAGE: &str = "Failed to generate completions";

    fn http_error_from_engine(code: u16) -> Result<(), anyhow::Error> {
        Err(HttpError {
            code,
            message: "custom error message".to_string(),
        })?
    }

    fn other_error_from_engine() -> Result<(), anyhow::Error> {
1096
        Err(ModelManagerError::ModelNotFound("foo".to_string()))?
1097
1098
    }

1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
    fn make_base_request() -> NvCreateResponse {
        NvCreateResponse {
            inner: CreateResponse {
                input: Input::Text("hello".into()),
                model: "test-model".into(),
                background: None,
                include: None,
                instructions: None,
                max_output_tokens: None,
                max_tool_calls: None,
                metadata: None,
                parallel_tool_calls: None,
                previous_response_id: None,
                prompt: None,
                reasoning: None,
                service_tier: None,
                store: None,
                stream: None,
                text: None,
                tool_choice: None,
                tools: None,
                truncation: None,
                user: None,
                temperature: None,
                top_logprobs: None,
                top_p: None,
            },
            nvext: None,
        }
    }

1130
1131
1132
    #[test]
    fn test_http_error_response_from_anyhow() {
        let err = http_error_from_engine(400).unwrap_err();
Ryan Olson's avatar
Ryan Olson committed
1133
        let (status, response) = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE);
1134
1135
1136
1137
1138
1139
1140
        assert_eq!(status, StatusCode::BAD_REQUEST);
        assert_eq!(response.error, "custom error message");
    }

    #[test]
    fn test_error_response_from_anyhow_out_of_range() {
        let err = http_error_from_engine(399).unwrap_err();
Ryan Olson's avatar
Ryan Olson committed
1141
        let (status, response) = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE);
1142
1143
1144
1145
        assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
        assert_eq!(response.error, "custom error message");

        let err = http_error_from_engine(500).unwrap_err();
Ryan Olson's avatar
Ryan Olson committed
1146
        let (status, response) = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE);
1147
1148
1149
1150
        assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
        assert_eq!(response.error, "custom error message");

        let err = http_error_from_engine(501).unwrap_err();
Ryan Olson's avatar
Ryan Olson committed
1151
        let (status, response) = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE);
1152
1153
1154
1155
1156
1157
1158
        assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
        assert_eq!(response.error, "custom error message");
    }

    #[test]
    fn test_other_error_response_from_anyhow() {
        let err = other_error_from_engine().unwrap_err();
Ryan Olson's avatar
Ryan Olson committed
1159
        let (status, response) = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE);
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
        assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
        assert_eq!(
            response.error,
            format!(
                "{}: {}",
                BACKUP_ERROR_MESSAGE,
                other_error_from_engine().unwrap_err()
            )
        );
    }
1170

1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
    #[test]
    fn test_service_overloaded_error_response_from_anyhow() {
        use dynamo_runtime::pipeline::error::PipelineError;

        let err: anyhow::Error = PipelineError::ServiceOverloaded(
            "All workers are busy, please retry later".to_string(),
        )
        .into();
        let (status, response) = ErrorMessage::from_anyhow(err, BACKUP_ERROR_MESSAGE);
        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(
            response.error,
            "Service temporarily unavailable: All workers are busy, please retry later"
        );
    }

1187
1188
1189
    #[test]
    fn test_validate_input_is_text_only_accepts_text() {
        let request = make_base_request();
1190
        let result = validate_response_input_is_text_only(&request);
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
        assert!(result.is_none());
    }

    #[test]
    fn test_validate_input_is_text_only_rejects_items() {
        let mut request = make_base_request();
        request.inner.input = Input::Items(vec![InputItem::Message(InputMessage {
            kind: Default::default(),
            role: ResponseRole::User,
            content: InputContent::TextInput("structured".into()),
        })]);
1202
        let result = validate_response_input_is_text_only(&request);
1203
1204
1205
1206
1207
1208
        assert!(result.is_some());
    }

    #[test]
    fn test_validate_unsupported_fields_accepts_clean_request() {
        let request = make_base_request();
1209
        let result = validate_response_unsupported_fields(&request);
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
        assert!(result.is_none());
    }

    #[test]
    fn test_validate_unsupported_fields_detects_flags() {
        #[allow(clippy::type_complexity)]
        let unsupported_cases: Vec<(&str, Box<dyn FnOnce(&mut CreateResponse)>)> = vec![
            ("background", Box::new(|r| r.background = Some(true))),
            (
                "include",
                Box::new(|r| r.include = Some(vec!["file_search_call.results".into()])),
            ),
            (
                "instructions",
                Box::new(|r| r.instructions = Some("System prompt".into())),
            ),
            ("max_tool_calls", Box::new(|r| r.max_tool_calls = Some(3))),
            ("metadata", Box::new(|r| r.metadata = Some(HashMap::new()))),
            (
                "parallel_tool_calls",
                Box::new(|r| r.parallel_tool_calls = Some(true)),
            ),
            (
                "previous_response_id",
                Box::new(|r| r.previous_response_id = Some("prev-id".into())),
            ),
            (
                "prompt",
                Box::new(|r| {
                    r.prompt = Some(PromptConfig {
                        id: "template-id".into(),
                        version: None,
                        variables: None,
                    })
                }),
            ),
            (
                "reasoning",
                Box::new(|r| r.reasoning = Some(Default::default())),
            ),
            (
                "service_tier",
                Box::new(|r| r.service_tier = Some(ServiceTier::Auto)),
            ),
            ("store", Box::new(|r| r.store = Some(true))),
            ("stream", Box::new(|r| r.stream = Some(true))),
            (
                "text",
                Box::new(|r| {
                    r.text = Some(TextConfig {
                        format: TextResponseFormat::Text,
                    })
                }),
            ),
            (
                "tool_choice",
                Box::new(|r| r.tool_choice = Some(ToolChoice::Mode(ToolChoiceMode::Required))),
            ),
            ("tools", Box::new(|r| r.tools = Some(vec![]))),
            (
                "truncation",
                Box::new(|r| r.truncation = Some(Truncation::Auto)),
            ),
            ("user", Box::new(|r| r.user = Some("user-id".into()))),
        ];

        for (field, set_field) in unsupported_cases {
            let mut req = make_base_request();
            (set_field)(&mut req.inner);
1279
            let result = validate_response_unsupported_fields(&req);
1280
1281
1282
            assert!(result.is_some(), "Expected rejection for `{field}`");
        }
    }
1283
1284
1285
1286
1287
1288
1289
1290
1291

    #[test]
    fn test_validate_chat_completion_required_fields_empty_messages() {
        let request = NvCreateChatCompletionRequest {
            inner: CreateChatCompletionRequest {
                model: "test-model".to_string(),
                messages: vec![],
                ..Default::default()
            },
1292
            common: Default::default(),
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
            nvext: None,
        };
        let result = validate_chat_completion_required_fields(&request);
        assert!(result.is_err());
        if let Err((status, error_response)) = result {
            assert_eq!(status, StatusCode::BAD_REQUEST);
            assert_eq!(
                error_response.error,
                "The 'messages' field cannot be empty. At least one message is required."
            );
        }
    }

    #[test]
    fn test_validate_chat_completion_required_fields_with_messages() {
        let request = NvCreateChatCompletionRequest {
            inner: CreateChatCompletionRequest {
                model: "test-model".to_string(),
                messages: vec![ChatCompletionRequestMessage::User(
                    ChatCompletionRequestUserMessage {
                        content: ChatCompletionRequestUserMessageContent::Text("Hello".to_string()),
                        name: None,
                    },
                )],
                ..Default::default()
            },
1319
            common: Default::default(),
1320
1321
1322
1323
1324
            nvext: None,
        };
        let result = validate_chat_completion_required_fields(&request);
        assert!(result.is_ok());
    }
1325
}