diff --git a/lib/llm/src/protocols/openai/chat_completions.rs b/lib/llm/src/protocols/openai/chat_completions.rs index e07af11084a5..e021ca8b9759 100644 --- a/lib/llm/src/protocols/openai/chat_completions.rs +++ b/lib/llm/src/protocols/openai/chat_completions.rs @@ -3,7 +3,7 @@ use std::collections::HashMap; -use dynamo_runtime::protocols::annotated::AnnotationsProvider; +use dynamo_runtime::protocols::annotated::{Annotated, AnnotationsProvider}; use serde::{Deserialize, Serialize}; use utoipa::ToSchema; use validator::Validate; @@ -30,8 +30,9 @@ pub use delta::DeltaGenerator; use dynamo_parsers::tool_calling::{ToolCallResponse, ToolCallResponseChunk}; use dynamo_protocols::types::{ - ChatCompletionMessageToolCall, ChatCompletionMessageToolCallChunk, FunctionCall, - FunctionCallStream, FunctionType, + ChatChoiceStream, ChatCompletionMessageContent, ChatCompletionMessageToolCall, + ChatCompletionMessageToolCallChunk, ChatCompletionStreamResponseDelta, FinishReason, + FunctionCall, FunctionCallStream, FunctionType, }; /// Map a parser-native [`ToolCallResponse`] onto the protocol/wire @@ -239,6 +240,45 @@ pub struct NvCreateChatCompletionStreamResponse { pub llm_metrics: Option, } +/// Build one synthetic stream choice from an existing response template. +/// +/// Both streaming tool-call paths use this constructor when an engine omits a +/// terminal choice. Accounting data belongs only on the usage chunk and must +/// not be copied onto the synthetic choice. +pub(super) fn stream_choice_chunk_from_template( + template: &NvCreateChatCompletionStreamResponse, + index: u32, + content: Option, + tool_calls: Option>, + finish_reason: Option, +) -> Annotated { + let mut response = template.clone(); + response.inner.usage = None; + response.llm_metrics = None; + #[allow(deprecated)] + let choice = ChatChoiceStream { + index, + delta: ChatCompletionStreamResponseDelta { + role: None, + content, + tool_calls, + function_call: None, + refusal: None, + reasoning_content: None, + }, + finish_reason, + logprobs: None, + }; + response.inner.choices = vec![choice]; + Annotated { + data: Some(response), + id: None, + event: None, + comment: None, + error: None, + } +} + /// Implements `NvExtProvider` for `NvCreateChatCompletionRequest`, /// providing access to NVIDIA-specific extensions. impl NvExtProvider for NvCreateChatCompletionRequest { diff --git a/lib/llm/src/protocols/openai/chat_completions/jail.rs b/lib/llm/src/protocols/openai/chat_completions/jail.rs index 543c19c694c7..1a6aa1697d31 100644 --- a/lib/llm/src/protocols/openai/chat_completions/jail.rs +++ b/lib/llm/src/protocols/openai/chat_completions/jail.rs @@ -17,12 +17,12 @@ use dynamo_parsers::tool_calling::{ }; use dynamo_runtime::protocols::annotated::Annotated; use futures::{Stream, StreamExt}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use uuid::Uuid; use crate::utils::{MarkerMatcher, MatchResult}; -use super::NvCreateChatCompletionStreamResponse; +use super::{NvCreateChatCompletionStreamResponse, stream_choice_chunk_from_template}; fn is_harmony_parser(parser: Option<&str>) -> bool { parser == Some("harmony") @@ -1461,17 +1461,37 @@ impl JailedStream { where S: Stream> + Send + 'static, { + let _ = named_tool_active; + let _ = &jail_mode; stream! { tokio::pin!(input_stream); let mut has_tool_calls_per_choice: HashMap = HashMap::new(); + // Choices that already received a finish_reason during the stream — used by + // the backstop below to avoid synthesizing a duplicate. + let mut terminated: HashSet = HashSet::new(); + // Last response, kept (with choices cleared) as a template for a synthesized + // finish_reason chunk when the stream ended without one. + let mut template: Option = None; + // Choices for which this post-processor has already emitted a synthetic + // terminal chunk. Tracking this per choice allows a later tool-call choice + // to terminate even if an earlier empty-choices chunk emitted nothing. + let mut synthesized: HashSet = HashSet::new(); while let Some(mut response) = input_stream.next().await { - // Track if any choice emitted tool calls + // Track if any choice emitted tool calls, and which already terminated. if let Some(ref data) = response.data { for choice in &data.inner.choices { if choice.delta.tool_calls.is_some() { has_tool_calls_per_choice.insert(choice.index, true); } + if choice.finish_reason.is_some() { + terminated.insert(choice.index); + } + } + { + let mut t = data.clone(); + t.inner.choices.clear(); + template = Some(t); } } @@ -1487,7 +1507,6 @@ impl JailedStream { // choice, finish_reason MUST be "tool_calls" — regardless of // whether tool_choice was "auto", "required", or a named // function. - let _ = named_tool_active; match &jail_mode { JailMode::MarkerBased => { if has_tool_calls { @@ -1506,8 +1525,68 @@ impl JailedStream { } } + // OpenAI stream ordering: the terminal finish_reason chunk must precede + // the usage-only chunk. When a chunk with no choices arrives (the + // frontend's compliance usage chunk, or any other empty-choices chunk) + // and tool-call choices are still missing a finish_reason, synthesize + // their terminal `ToolCalls` chunks *before* yielding this one. + let is_empty_choices = response + .data + .as_ref() + .is_some_and(|d| d.inner.choices.is_empty()); + if is_empty_choices && let Some(template) = &template { + let mut indices: Vec<_> = has_tool_calls_per_choice + .iter() + .filter_map(|(index, has)| { + (*has && !terminated.contains(index) && !synthesized.contains(index)) + .then_some(*index) + }) + .collect(); + indices.sort_unstable(); + for index in indices { + yield stream_choice_chunk_from_template( + template, + index, + None, + None, + Some(FinishReason::ToolCalls), + ); + synthesized.insert(index); + } + } + yield response; } + + // Backstop: the stream ended without a finish_reason AND without an + // empty-choices/usage chunk to anchor the synthesized terminal chunks + // before (e.g. the engine dropped the terminal signal and the frontend + // never emitted a usage chunk). Emit one trailing `ToolCalls` chunk per + // tool-call choice that never received a finish_reason. Strict OpenAI + // clients wait for a non-null finish_reason before considering a tool call + // complete; without this they hang until their client-side timeout. + // Choices that never emitted tool calls are left alone — there + // is no signal to invent a finish_reason from for text-only output. + if let Some(template) = template { + let mut indices: Vec<_> = has_tool_calls_per_choice + .iter() + .filter_map(|(index, has)| { + (*has && !terminated.contains(index) && !synthesized.contains(index)) + .then_some(*index) + }) + .collect(); + indices.sort_unstable(); + for index in indices { + yield stream_choice_chunk_from_template( + &template, + index, + None, + None, + Some(FinishReason::ToolCalls), + ); + synthesized.insert(index); + } + } } } } @@ -1775,6 +1854,96 @@ mod tests { } } + /// A usage-only chunk (empty `choices`, a usage object) — the frontend's + /// OpenAI-compliance terminal chunk. Used to test that the synthesized + /// `ToolCalls` finish_reason chunk is emitted *before* this one. + fn usage_only_chunk() -> Annotated { + Annotated { + data: Some(NvCreateChatCompletionStreamResponse { + inner: CreateChatCompletionStreamResponse { + id: "id-42".to_string(), + object: "chat.completion.chunk".to_string(), + created: 0, + model: "test-model".to_string(), + choices: vec![], + usage: Some(dynamo_protocols::types::CompletionUsage { + prompt_tokens: 10, + completion_tokens: 5, + total_tokens: 15, + prompt_tokens_details: None, + completion_tokens_details: None, + }), + service_tier: None, + system_fingerprint: None, + }, + nvext: None, + llm_metrics: Some(crate::protocols::common::metrics::LLMMetricAnnotation { + input_tokens: 10, + output_tokens: 5, + chunk_tokens: 0, + cached_tokens: None, + prefill_worker_id: None, + prefill_dp_rank: None, + prefill_worker_type: None, + decode_worker_id: None, + decode_dp_rank: None, + decode_worker_type: None, + tokenize_latency: None, + detokenize_total_latency: None, + detokenize_count: None, + }), + }), + id: None, + event: None, + comment: None, + error: None, + } + } + + /// Build one data chunk whose choices have already emitted tool-call deltas. + fn tool_call_choices_chunk(indices: &[u32]) -> Annotated { + let mut chunk = text_chunk(""); + let data = chunk.data.as_mut().expect("tool-call response data"); + #[allow(deprecated)] + { + data.inner.choices = indices + .iter() + .map(|index| ChatChoiceStream { + index: *index, + delta: ChatCompletionStreamResponseDelta { + role: Some(Role::Assistant), + content: None, + tool_calls: Some(vec![ChatCompletionMessageToolCallChunk { + index: 0, + id: Some(format!("call-{index}")), + r#type: Some(FunctionType::Function), + function: Some(FunctionCallStream { + name: Some(format!("tool_{index}")), + arguments: Some("{}".to_string()), + }), + }]), + function_call: None, + refusal: None, + reasoning_content: None, + }, + finish_reason: None, + logprobs: None, + }) + .collect(); + } + chunk + } + + fn heartbeat() -> Annotated { + Annotated { + data: None, + id: None, + event: None, + comment: Some(vec!["heartbeat".to_string()]), + error: None, + } + } + /// Collect all emitted tool calls from the jailed stream output fn collect_tool_calls( responses: &[Annotated], @@ -2079,4 +2248,209 @@ mod tests { all_text ); } + + /// The last `finish_reason` a client sees across the stream. `None` if the + /// stream never carried one (the missing-finish-reason hang condition). + fn final_finish_reason( + responses: &[Annotated], + ) -> Option { + responses + .iter() + .filter_map(|r| r.data.as_ref()) + .flat_map(|d| d.inner.choices.iter()) + .filter_map(|c| c.finish_reason) + .next_back() + } + + // Missing-finish-reason regression: when the engine emits a complete tool call + // but the stream ends without any finish_reason chunk (speculative decoding + // folded EOS into content, or the terminal signal was dropped), a strict + // OpenAI client waits for a non-null finish_reason and hangs until its timeout. + // The jail path's finalize() emits the tool call with the absent upstream + // finish_reason; fix_finish_reason's end-of-stream path must synthesize + // `ToolCalls` so the client gets a terminal signal. + #[tokio::test] + async fn jail_synthesizes_tool_calls_finish_reason_when_stream_lacks_one() { + let jail = JailedStream::builder().tool_call_parser("hermes").build(); + + let chunks = vec![text_chunk( + "\n{\"name\": \"get_weather\", \"arguments\": {\"location\": \"SF\"}}\n", + )]; + + let input_stream = Box::pin(stream::iter(chunks)); + let output_stream = jail.apply_with_finish_reason(input_stream); + + let responses: Vec<_> = output_stream.collect().await; + let tool_calls = collect_tool_calls(&responses); + assert!( + !tool_calls.is_empty(), + "expected the hermes tool call to be parsed: {tool_calls:?}" + ); + assert_eq!(tool_calls[0].0, "get_weather"); + assert_eq!( + final_finish_reason(&responses), + Some(FinishReason::ToolCalls), + "backstop must synthesize ToolCalls when the stream ended without a finish_reason" + ); + } + + // Text-only corollary: a text-only stream that ends without a finish_reason + // must not get a synthesized one. There is no signal to invent a + // finish_reason from when no tool call was emitted. + #[tokio::test] + async fn jail_does_not_synthesize_finish_reason_for_text_only_stream() { + let jail = JailedStream::builder().tool_call_parser("hermes").build(); + + let chunks = vec![text_chunk("hello world"), text_chunk("")]; + + let input_stream = Box::pin(stream::iter(chunks)); + let output_stream = jail.apply_with_finish_reason(input_stream); + + let responses: Vec<_> = output_stream.collect().await; + let tool_calls = collect_tool_calls(&responses); + assert!( + tool_calls.is_empty(), + "no tool calls expected: {tool_calls:?}" + ); + assert_eq!( + final_finish_reason(&responses), + None, + "text-only stream with no upstream finish_reason must not get a synthetic one" + ); + } + + // Usage-ordering regression: a tool call is followed by a usage-only chunk, + // with no finish_reason chunk from the engine. The + // synthesized `ToolCalls` terminal chunk must be emitted *before* the + // usage-only chunk (OpenAI stream ordering — the terminal finish_reason + // precedes usage). This mirrors the production stream ordering. + #[tokio::test] + async fn jail_synthesizes_tool_calls_before_usage_only_chunk() { + let jail = JailedStream::builder().tool_call_parser("hermes").build(); + + let chunks = vec![ + heartbeat(), + text_chunk( + "\n{\"name\": \"get_weather\", \"arguments\": {\"location\": \"SF\"}}\n", + ), + usage_only_chunk(), + ]; + + let input_stream = Box::pin(stream::iter(chunks)); + let output_stream = jail.apply_with_finish_reason(input_stream); + + let responses: Vec<_> = output_stream.collect().await; + assert_eq!( + responses + .first() + .and_then(|response| response.comment.clone()), + Some(vec!["heartbeat".to_string()]), + "leading non-data annotations must pass through unchanged" + ); + let tool_calls = collect_tool_calls(&responses); + assert!( + !tool_calls.is_empty(), + "expected the hermes tool call: {tool_calls:?}" + ); + assert_eq!(tool_calls[0].0, "get_weather"); + assert_eq!( + final_finish_reason(&responses), + Some(FinishReason::ToolCalls), + "a synthesized ToolCalls terminal chunk must be present" + ); + + // The ToolCalls terminal chunk must precede the usage-only chunk. + let finish_pos = responses.iter().position(|r| { + r.data.as_ref().is_some_and(|d| { + d.inner + .choices + .iter() + .any(|c| c.finish_reason == Some(FinishReason::ToolCalls)) + }) + }); + let usage_pos = responses.iter().position(|r| { + r.data + .as_ref() + .is_some_and(|d| d.inner.usage.is_some() && d.inner.choices.is_empty()) + }); + let finish_pos = finish_pos.expect("no ToolCalls chunk emitted"); + let usage_pos = usage_pos.expect("no usage-only chunk in output"); + assert!( + finish_pos < usage_pos, + "ToolCalls chunk at {finish_pos} must precede the usage chunk at {usage_pos}" + ); + let finish_data = responses[finish_pos] + .data + .as_ref() + .expect("ToolCalls chunk has no response data"); + assert!( + finish_data.inner.usage.is_none(), + "synthesized ToolCalls chunk must not repeat usage data" + ); + assert!( + finish_data.llm_metrics.is_none(), + "synthesized ToolCalls chunk must not repeat LLM metrics" + ); + } + + // An empty-choices chunk can precede tool deltas (for example, a metadata + // response). It must not disable later synthesis. When several choices then + // emit tool calls, their terminal chunks must be ordered by choice index. + #[tokio::test] + async fn jail_synthesizes_late_tool_choices_in_index_order() { + let chunks = vec![ + usage_only_chunk(), + tool_call_choices_chunk(&[2, 0, 1]), + usage_only_chunk(), + ]; + + let responses: Vec<_> = + JailedStream::fix_finish_reason(stream::iter(chunks), JailMode::MarkerBased, false) + .collect() + .await; + + let usage_positions: Vec<_> = responses + .iter() + .enumerate() + .filter_map(|(position, response)| { + response + .data + .as_ref() + .is_some_and(|data| data.inner.choices.is_empty() && data.inner.usage.is_some()) + .then_some(position) + }) + .collect(); + assert_eq!( + usage_positions.len(), + 2, + "both empty-choices chunks must pass through" + ); + + let terminals: Vec<_> = responses + .iter() + .enumerate() + .flat_map(|(position, response)| { + response.data.iter().flat_map(move |data| { + data.inner.choices.iter().filter_map(move |choice| { + (choice.finish_reason == Some(FinishReason::ToolCalls)) + .then_some((position, choice.index)) + }) + }) + }) + .collect(); + assert_eq!( + terminals + .iter() + .map(|(_, index)| *index) + .collect::>(), + vec![0, 1, 2], + "synthetic terminal chunks must be deterministic" + ); + assert!( + terminals.iter().all(|(position, _)| { + usage_positions[0] < *position && *position < usage_positions[1] + }), + "terminal chunks must follow the early empty response and precede the final usage response" + ); + } } diff --git a/lib/llm/src/protocols/openai/chat_completions/tool_parser_v2.rs b/lib/llm/src/protocols/openai/chat_completions/tool_parser_v2.rs index 9ce7c12a02ff..4cdf482b8e50 100644 --- a/lib/llm/src/protocols/openai/chat_completions/tool_parser_v2.rs +++ b/lib/llm/src/protocols/openai/chat_completions/tool_parser_v2.rs @@ -33,7 +33,7 @@ use dynamo_parsers::tool_calling::{ }; use dynamo_parsers_v2::{Tool as ToolV2, ToolCallDelta, ToolParser, create_tool_parser_for_family}; -use super::NvCreateChatCompletionStreamResponse; +use super::{NvCreateChatCompletionStreamResponse, stream_choice_chunk_from_template}; /// Tool-call families with a `dynamo-parsers-v2` parser wired into both the batch and /// the streaming path. Must stay a subset of the families @@ -163,6 +163,62 @@ impl ChoiceState { } } +/// Finish every choice that has not received an upstream finish reason. This is +/// called before a usage-only chunk when one exists, with EOF as a fallback. +fn finish_unterminated_choices( + states: &mut HashMap, + finished: &mut HashSet, + tool_emitted: &mut HashSet, + template: &NvCreateChatCompletionStreamResponse, +) -> Vec> { + let mut indices: Vec<_> = states + .keys() + .copied() + .filter(|index| !finished.contains(index)) + .collect(); + indices.sort_unstable(); + + let mut responses = Vec::new(); + for index in indices { + finished.insert(index); + let state = states + .get_mut(&index) + .expect("choice index came from parser state map"); + let result = match state.parser.finish() { + Ok(result) => result, + Err(error) => { + tracing::warn!(error = %error, choice_index = index, "v2 stream finish failed"); + dynamo_parsers_v2::ToolParseResult::default() + } + }; + let tool_calls = state.emit_chunks(result.calls); + if tool_calls.is_some() { + tool_emitted.insert(index); + } + // A choice that produced tool calls during the stream must terminate + // with `ToolCalls` even when the backend never sent a finish_reason. + // Text-only output without an upstream finish reason stays `None`. + let finish_reason = if tool_emitted.contains(&index) { + Some(FinishReason::ToolCalls) + } else { + None + }; + let content = (!result.normal_text.is_empty()) + .then_some(ChatCompletionMessageContent::Text(result.normal_text)); + if content.is_none() && tool_calls.is_none() && finish_reason.is_none() { + continue; + } + responses.push(stream_choice_chunk_from_template( + template, + index, + content, + tool_calls, + finish_reason, + )); + } + responses +} + /// Streaming path: replace the jail with the `family` v2 parser. Each upstream text /// delta is pushed into the parser; the parser's `normal_text` becomes the emitted /// content and its tool-call deltas become OpenAI tool-call chunks. The jail is never @@ -213,6 +269,7 @@ where t.inner.choices.clear(); template = Some(t); } + let is_empty_choices = chat_response.inner.choices.is_empty(); for choice in chat_response.inner.choices.iter_mut() { let state = states.entry(choice.index).or_insert_with(|| { @@ -282,47 +339,36 @@ where } } + // OpenAI stream ordering requires a terminal finish_reason before the + // usage-only chunk. Finish every unterminated choice before yielding an + // empty-choices response; EOF below remains the fallback when no such + // response arrives. + if is_empty_choices && let Some(template) = &template { + for terminal in finish_unterminated_choices( + &mut states, + &mut finished, + &mut tool_emitted, + template, + ) { + yield terminal; + } + } + yield response; } // Backstop: the stream ended without a finish_reason for some choice. Flush - // each unfinished parser; emit a trailing chunk only when it yields output. - if let Some(template) = template { - for (index, state) in states.iter_mut() { - if finished.contains(index) { - continue; - } - let Ok(result) = state.parser.finish() else { - continue; - }; - let tool_calls = state.emit_chunks(result.calls); - if result.normal_text.is_empty() && tool_calls.is_none() { - continue; - } - let mut response = template.clone(); - #[allow(deprecated)] - let choice = dynamo_protocols::types::ChatChoiceStream { - index: *index, - delta: dynamo_protocols::types::ChatCompletionStreamResponseDelta { - role: None, - content: (!result.normal_text.is_empty()) - .then_some(ChatCompletionMessageContent::Text(result.normal_text)), - tool_calls, - function_call: None, - refusal: None, - reasoning_content: None, - }, - finish_reason: None, - logprobs: None, - }; - response.inner.choices = vec![choice]; - yield Annotated { - data: Some(response), - id: None, - event: None, - comment: None, - error: None, - }; + // each unfinished parser; emit a trailing chunk when the flush yields output + // or when the choice already emitted tool calls and still needs a terminal + // `ToolCalls` reason. + if let Some(template) = &template { + for terminal in finish_unterminated_choices( + &mut states, + &mut finished, + &mut tool_emitted, + template, + ) { + yield terminal; } } } @@ -332,10 +378,26 @@ where mod tests { use super::*; use dynamo_protocols::types::{ - ChatChoiceStream, ChatCompletionStreamResponseDelta, FinishReason, Role, + ChatChoiceStream, ChatCompletionStreamResponseDelta, CompletionUsage, FinishReason, Role, }; use futures::stream; + struct FinishErrorParser; + + impl ToolParser for FinishErrorParser { + fn create(_tools: &[ToolV2]) -> anyhow::Result> { + Ok(Box::new(Self)) + } + + fn push(&mut self, _chunk: &str) -> anyhow::Result { + Ok(dynamo_parsers_v2::ToolParseResult::default()) + } + + fn finish(&mut self) -> anyhow::Result { + anyhow::bail!("intentional finish failure") + } + } + const QWEN3_GET_WEATHER: &str = "\n\n\nParis\n\n\n"; // DeepSeek-V4 DSML: one get_weather(location="NYC") call. The `｜` glyphs are the @@ -379,6 +441,35 @@ mod tests { } } + fn usage_chunk() -> Annotated { + let mut chunk = chunk("", false); + let data = chunk.data.as_mut().expect("usage chunk response data"); + data.inner.choices.clear(); + data.inner.usage = Some(CompletionUsage { + prompt_tokens: 10, + completion_tokens: 5, + total_tokens: 15, + prompt_tokens_details: None, + completion_tokens_details: None, + }); + data.llm_metrics = Some(crate::protocols::common::metrics::LLMMetricAnnotation { + input_tokens: 10, + output_tokens: 5, + chunk_tokens: 0, + cached_tokens: None, + prefill_worker_id: None, + prefill_dp_rank: None, + prefill_worker_type: None, + decode_worker_id: None, + decode_dp_rank: None, + decode_worker_type: None, + tokenize_latency: None, + detokenize_total_latency: None, + detokenize_count: None, + }); + chunk + } + /// Reassemble the streamed tool-call deltas into (name, arguments) per index and /// collect all emitted content, mirroring how an OpenAI client reconstructs a /// streamed tool call. @@ -535,4 +626,139 @@ mod tests { "finish_reason must flip Stop->ToolCalls when tool calls are emitted" ); } + + // Missing-finish-reason regression: the stream emits a complete tool call but + // ends without any finish_reason chunk (e.g. speculative decoding folded EOS + // into content, or the engine dropped the terminal signal). A strict OpenAI + // client waits for a non-null finish_reason before considering the tool call + // complete; the end-of-stream backstop must synthesize `ToolCalls` so the + // client doesn't hang until its timeout. + #[tokio::test] + async fn qwen3_bypass_synthesizes_tool_calls_when_stream_lacks_finish_reason() { + // Same call as the incremental test, but the final chunk carries NO + // finish_reason — the stream simply ends after the tool markup. + let mut chunks: Vec<_> = QWEN3_GET_WEATHER + .as_bytes() + .chunks(8) + .map(|b| chunk(std::str::from_utf8(b).unwrap(), false)) + .collect(); + // A usage-only chunk arrives without any terminating choice. + chunks.push(usage_chunk()); + + let out: Vec<_> = apply_stream(stream::iter(chunks), None, "qwen3_coder".to_string()) + .collect::>() + .await; + + let (calls, _content) = reassemble(&out); + assert_eq!(calls.len(), 1, "expected exactly one tool call: {calls:?}"); + assert_eq!(calls[0].0, "get_weather"); + let args: serde_json::Value = serde_json::from_str(&calls[0].1).unwrap(); + assert_eq!(args["location"], "Paris"); + assert_eq!( + final_finish_reason(&out), + Some(FinishReason::ToolCalls), + "backstop must synthesize ToolCalls when the stream ended without a finish_reason" + ); + let finish_positions: Vec<_> = out + .iter() + .enumerate() + .filter_map(|(position, response)| { + response.data.as_ref().and_then(|data| { + data.inner + .choices + .iter() + .any(|choice| choice.finish_reason == Some(FinishReason::ToolCalls)) + .then_some(position) + }) + }) + .collect(); + assert_eq!( + finish_positions.len(), + 1, + "expected exactly one synthesized finish chunk" + ); + let usage_position = + out.iter() + .position(|response| { + response.data.as_ref().is_some_and(|data| { + data.inner.choices.is_empty() && data.inner.usage.is_some() + }) + }) + .expect("usage-only response"); + assert!( + finish_positions[0] < usage_position, + "synthesized finish chunk must precede usage" + ); + let terminal = out[finish_positions[0]] + .data + .as_ref() + .expect("synthesized terminal response"); + assert!( + terminal.inner.usage.is_none(), + "synthesized terminal chunk must not repeat usage" + ); + assert!( + terminal.llm_metrics.is_none(), + "synthesized terminal chunk must not repeat LLM metrics" + ); + } + + // Text-only corollary: when the stream ends without a finish_reason and no + // tool call was emitted, the backstop must not invent a finish_reason. There + // is no signal to synthesize one from. A trailing content chunk may be + // emitted, but its finish_reason stays None. + #[tokio::test] + async fn qwen3_bypass_does_not_synthesize_finish_reason_for_text_only_stream() { + let chunks = vec![chunk("hello world", false), chunk("", false)]; + + let out: Vec<_> = apply_stream(stream::iter(chunks), None, "qwen3_coder".to_string()) + .collect::>() + .await; + + let (calls, _content) = reassemble(&out); + assert!(calls.is_empty(), "no tool calls expected: {calls:?}"); + assert_eq!( + final_finish_reason(&out), + None, + "text-only stream with no upstream finish_reason must not get a synthetic one" + ); + } + + #[test] + fn finish_error_still_terminates_a_choice_that_emitted_tools() { + let mut states = HashMap::from([( + 3, + ChoiceState { + parser: Box::new(FinishErrorParser), + opened: HashSet::new(), + }, + )]); + let mut finished = HashSet::new(); + let mut tool_emitted = HashSet::from([3]); + let template = usage_chunk().data.expect("usage response data"); + + let responses = + finish_unterminated_choices(&mut states, &mut finished, &mut tool_emitted, &template); + + assert_eq!( + responses.len(), + 1, + "the choice still needs a terminal chunk" + ); + let response = responses[0].data.as_ref().expect("terminal response data"); + assert!( + response.inner.usage.is_none(), + "terminal chunk must not repeat usage" + ); + assert!( + response.llm_metrics.is_none(), + "terminal chunk must not repeat LLM metrics" + ); + assert_eq!(response.inner.choices.len(), 1); + assert_eq!(response.inner.choices[0].index, 3); + assert_eq!( + response.inner.choices[0].finish_reason, + Some(FinishReason::ToolCalls) + ); + } } diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index 2619009802d7..4a668bd2446a 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -1106,6 +1106,29 @@ mod tests { assert!(end_types.contains(&"response.completed".to_string())); } + #[test] + fn test_function_call_finish_reason_closes_tool_call() { + let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); + let _ = conv.emit_start_events(); + + let _ = conv.process_chunk(&tool_call_chunk( + 0, + Some("call-1"), + Some("get_weather"), + Some("{\"city\":\"SF\"}"), + )); + + let finish_types = + event_types(&conv.process_chunk(&finish_chunk(FinishReason::FunctionCall))); + assert_eq!( + finish_types, + vec![ + "response.function_call_arguments.done".to_string(), + "response.output_item.done".to_string(), + ] + ); + } + #[test] fn test_identity_only_tool_call_is_emitted_and_finished() { let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); diff --git a/lib/llm/tests/common/http_harness.rs b/lib/llm/tests/common/http_harness.rs index 9c187b616d49..76f3eb2f4a5c 100644 --- a/lib/llm/tests/common/http_harness.rs +++ b/lib/llm/tests/common/http_harness.rs @@ -139,6 +139,7 @@ pub async fn load_sse_fixture(path: impl AsRef) -> Result