diff --git a/lib/components/fabro-llm/src/codec/openai_compatible/request.rs b/lib/components/fabro-llm/src/codec/openai_compatible/request.rs index 1ec6dc8fb..f6bbf3b16 100644 --- a/lib/components/fabro-llm/src/codec/openai_compatible/request.rs +++ b/lib/components/fabro-llm/src/codec/openai_compatible/request.rs @@ -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.` 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] diff --git a/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs b/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs index c8facd920..4327176fc 100644 --- a/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs +++ b/lib/components/fabro-llm/tests/it/wire/openai_compatible.rs @@ -560,7 +560,8 @@ async fn stream_text_happy_path_capture() -> (WireCapture, Vec, #[serde(default)] pub stream: bool, + stream_options: Option, pub tools: Option>, pub tool_choice: Option, pub response_format: Option, pub stop: Option, } +#[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 = self .messages diff --git a/test/twin/openai/src/sse.rs b/test/twin/openai/src/sse.rs index eb49c3f47..2d30a0f88 100644 --- a/test/twin/openai/src/sse.rs +++ b/test/twin/openai/src/sse.rs @@ -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 { +pub fn chat_sse_response( + plan: &ResponsePlan, + include_usage: bool, + transport: TransportOptions, +) -> Response { 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()); } diff --git a/test/twin/openai/tests/chat_completions_contract.rs b/test/twin/openai/tests/chat_completions_contract.rs index 6fe4c27fe..e17369c42 100644 --- a/test/twin/openai/tests/chat_completions_contract.rs +++ b/test/twin/openai/tests/chat_completions_contract.rs @@ -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::(&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");