test(llm): model streamed usage in OpenAI twin

This commit is contained in:
Bryan Helmkamp 2026-07-24 17:47:05 -04:00
parent 0169725b4e
commit 53a92f75f4
No known key found for this signature in database
6 changed files with 77 additions and 45 deletions

View file

@ -177,13 +177,10 @@ mod tests {
tool_choice: None,
response_format: None,
stream: Some(true),
stream_options: Some(StreamOptions {
include_usage: true,
}),
stream_options: None,
};
let json = serde_json::to_value(&req).unwrap();
assert_eq!(json["stream"], true);
assert_eq!(json["stream_options"]["include_usage"], true);
let req_no_stream = ApiRequest {
model: "test".into(),
@ -201,44 +198,6 @@ mod tests {
};
let json_no_stream = serde_json::to_value(&req_no_stream).unwrap();
assert!(json_no_stream.get("stream").is_none());
assert!(json_no_stream.get("stream_options").is_none());
}
/// Chat Completions only emits the trailing usage chunk when the request
/// opts in; without it streamed responses report zero tokens and cost
/// estimation silently produces $0.
#[test]
fn encode_opts_into_streaming_usage_when_streaming() {
let request = minimal_request();
let body = encode_body(&request, "kimi", true);
assert_eq!(body["stream"], true);
assert_eq!(body["stream_options"]["include_usage"], true);
}
#[test]
fn encode_omits_stream_options_when_not_streaming() {
let request = minimal_request();
let body = encode_body(&request, "kimi", false);
assert!(body.get("stream_options").is_none());
}
/// The `provider_options.<name>` merge runs after the body is built, so a
/// caller pointed at a gateway that rejects the field can still turn it
/// off.
#[test]
fn provider_options_can_override_stream_options() {
let mut request = minimal_request();
request.provider_options = Some(serde_json::json!({
"kimi": { "stream_options": serde_json::Value::Null }
}));
let body = encode_body(&request, "kimi", true);
assert_eq!(body["stream_options"], serde_json::Value::Null);
}
#[test]

View file

@ -560,7 +560,8 @@ async fn stream_text_happy_path_capture() -> (WireCapture, Vec<serde_json::Value
stream_capture(&base_request(MODEL), &sse).await
}
/// The captured request pins the stream flag on the wire.
/// The captured request pins the streaming request shape, including the usage
/// opt-in required for the trailing usage chunk.
#[tokio::test]
async fn stream_text_happy_path_request() {
let (capture, _) = stream_text_happy_path_capture().await;

View file

@ -43,7 +43,12 @@ pub async fn create_chat_completion(
}
if request.stream {
chat_sse_response(&success.plan, success.transport).into_response()
chat_sse_response(
&success.plan,
request.include_stream_usage(),
success.transport,
)
.into_response()
} else {
Json(success.plan.chat_completions_json()).into_response()
}

View file

@ -534,13 +534,27 @@ pub struct ChatCompletionsRequest {
pub max_tokens: Option<i64>,
#[serde(default)]
pub stream: bool,
stream_options: Option<ChatStreamOptions>,
pub tools: Option<Vec<Value>>,
pub tool_choice: Option<Value>,
pub response_format: Option<ChatResponseFormat>,
pub stop: Option<Value>,
}
#[derive(Clone, Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct ChatStreamOptions {
#[serde(default)]
include_usage: bool,
}
impl ChatCompletionsRequest {
pub fn include_stream_usage(&self) -> bool {
self.stream_options
.as_ref()
.is_some_and(|options| options.include_usage)
}
pub fn extract_user_text(&self) -> String {
let pieces: Vec<String> = self
.messages

View file

@ -236,7 +236,11 @@ pub fn responses_sse_response(plan: &ResponsePlan, transport: TransportOptions)
stream_response(events, transport)
}
pub fn chat_sse_response(plan: &ResponsePlan, transport: TransportOptions) -> Response<Body> {
pub fn chat_sse_response(
plan: &ResponsePlan,
include_usage: bool,
transport: TransportOptions,
) -> Response<Body> {
let mut events = Vec::new();
let content = plan.chat_content();
events.push(chat_chunk(&json!({
@ -321,6 +325,16 @@ pub fn chat_sse_response(plan: &ResponsePlan, transport: TransportOptions) -> Re
"finish_reason": if plan.tool_calls.is_empty() { "stop" } else { "tool_calls" },
}]
})));
if include_usage {
events.push(chat_chunk(&json!({
"id": format!("chatcmpl_{}", plan.id),
"object": "chat.completion.chunk",
"created": plan.created,
"model": plan.model,
"choices": [],
"usage": plan.usage.chat_completions_json(),
})));
}
events.push("data: [DONE]\n\n".to_owned());
}

View file

@ -50,9 +50,48 @@ async fn chat_completions_stream_uses_same_canonical_plan() {
assert_eq!(status, 200);
assert!(joined.contains("\"content\":\"deterministic: stream same plan\""));
assert!(!joined.contains("\"usage\""));
assert!(joined.contains("data: [DONE]"));
}
#[tokio::test]
async fn chat_completions_stream_includes_usage_when_requested() {
let server = common::spawn_server().await.expect("server should start");
let (status, chunks) = server
.post_chat_stream(json!({
"model": "gpt-test",
"messages": [{ "role": "user", "content": "stream with usage" }],
"stream": true,
"stream_options": { "include_usage": true }
}))
.await;
assert_eq!(status, 200);
let transcript =
common::parse_sse_transcript(chunks.join("").as_bytes()).expect("valid SSE transcript");
let usage_chunk = transcript
.events
.iter()
.filter(|event| event.data != "[DONE]")
.map(|event| {
serde_json::from_str::<serde_json::Value>(&event.data).expect("valid JSON chunk")
})
.find(|chunk| chunk.get("usage").is_some())
.expect("trailing usage chunk");
assert_eq!(usage_chunk["choices"], json!([]));
assert_eq!(
usage_chunk["usage"],
json!({
"prompt_tokens": 3,
"completion_tokens": 5,
"total_tokens": 8
})
);
assert!(transcript.done);
}
#[tokio::test]
async fn chat_completions_accepts_supported_openai_compatible_fields() {
let server = common::spawn_server().await.expect("server should start");