From 48d19056cc416eb13914c6b0a3d5fcdce9a7d9fc Mon Sep 17 00:00:00 2001 From: Ishan Dhanani Date: Fri, 27 Mar 2026 13:19:04 +0000 Subject: [PATCH 1/9] fix: improve responses stream conversion for tool calls --- .../openai/responses/stream_converter.rs | 304 +++++++++++++----- 1 file changed, 217 insertions(+), 87 deletions(-) diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index a3745fec1b68..d81b311dbc81 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -27,6 +27,8 @@ use uuid::Uuid; use dynamo_async_openai::types::ChatCompletionMessageContent; use super::ResponseParams; +use super::parse_tool_call_text; +use super::strip_tool_call_text; use crate::protocols::openai::chat_completions::NvCreateChatCompletionStreamResponse; /// State machine that converts a chat completion stream into Responses API events. @@ -40,9 +42,11 @@ pub struct ResponseStreamConverter { message_item_id: String, message_started: bool, message_output_index: u32, + raw_text: String, accumulated_text: String, // Function call tracking function_call_items: Vec, + fallback_tool_calls_emitted: usize, // Output index counter next_output_index: u32, // Usage stats from the backend's final chunk @@ -77,8 +81,10 @@ impl ResponseStreamConverter { message_item_id: format!("msg_{}", Uuid::new_v4().simple()), message_started: false, message_output_index: 0, + raw_text: String::new(), accumulated_text: String::new(), function_call_items: Vec::new(), + fallback_tool_calls_emitted: 0, next_output_index: 0, usage: None, } @@ -90,6 +96,193 @@ impl ResponseStreamConverter { seq } + fn append_raw_text(&mut self, content: &str) { + if content.is_empty() || content == self.raw_text { + return; + } + + if content.starts_with(&self.raw_text) { + self.raw_text = content.to_string(); + } else { + self.raw_text.push_str(content); + } + } + + fn ensure_message_started(&mut self, events: &mut Vec>) { + if self.message_started { + return; + } + + self.message_started = true; + self.message_output_index = self.next_output_index; + let output_index = self.message_output_index; + self.next_output_index += 1; + + let item_added = + ResponseStreamEvent::ResponseOutputItemAdded(ResponseOutputItemAddedEvent { + sequence_number: self.next_seq(), + output_index, + item: OutputItem::Message(OutputMessage { + id: Some(self.message_item_id.clone()), + content: vec![], + role: AssistantRole::Assistant, + status: Some(OutputStatus::InProgress), + }), + }); + events.push(make_sse_event(&item_added)); + + let part_added = + ResponseStreamEvent::ResponseContentPartAdded(ResponseContentPartAddedEvent { + sequence_number: self.next_seq(), + item_id: self.message_item_id.clone(), + output_index, + content_index: 0, + part: OutputContent::OutputText(OutputTextContent { + text: String::new(), + annotations: vec![], + logprobs: Some(vec![]), + }), + }); + events.push(make_sse_event(&part_added)); + } + + fn emit_visible_text_delta( + &mut self, + events: &mut Vec>, + visible_text: &str, + ) { + let Some(delta) = visible_text.strip_prefix(&self.accumulated_text) else { + return; + }; + if delta.is_empty() { + return; + } + + self.ensure_message_started(events); + self.accumulated_text = visible_text.to_string(); + + let text_delta = ResponseStreamEvent::ResponseOutputTextDelta(ResponseTextDeltaEvent { + sequence_number: self.next_seq(), + item_id: self.message_item_id.clone(), + output_index: self.message_output_index, + content_index: 0, + delta: delta.to_string(), + logprobs: Some(vec![]), + }); + events.push(make_sse_event(&text_delta)); + } + + fn emit_fallback_tool_call( + &mut self, + events: &mut Vec>, + name: String, + arguments: String, + ) { + let item_id = format!("fc_{}", Uuid::new_v4().simple()); + let call_id = format!("call_{}", Uuid::new_v4().simple()); + let output_index = self.next_output_index; + self.next_output_index += 1; + + self.function_call_items.push(FunctionCallState { + item_id: item_id.clone(), + call_id: call_id.clone(), + name: name.clone(), + accumulated_args: arguments.clone(), + output_index, + started: true, + done: true, + }); + + let item_added = + ResponseStreamEvent::ResponseOutputItemAdded(ResponseOutputItemAddedEvent { + sequence_number: self.next_seq(), + output_index, + item: OutputItem::FunctionCall(FunctionToolCall { + id: Some(item_id.clone()), + call_id: call_id.clone(), + name: name.clone(), + arguments: String::new(), + status: Some(OutputStatus::InProgress), + }), + }); + events.push(make_sse_event(&item_added)); + + let args_done = ResponseStreamEvent::ResponseFunctionCallArgumentsDone( + ResponseFunctionCallArgumentsDoneEvent { + sequence_number: self.next_seq(), + item_id: item_id.clone(), + output_index, + arguments: arguments.clone(), + name: Some(name.clone()), + }, + ); + events.push(make_sse_event(&args_done)); + + let item_done = ResponseStreamEvent::ResponseOutputItemDone(ResponseOutputItemDoneEvent { + sequence_number: self.next_seq(), + output_index, + item: OutputItem::FunctionCall(FunctionToolCall { + id: Some(item_id), + call_id, + name, + arguments, + status: Some(OutputStatus::Completed), + }), + }); + events.push(make_sse_event(&item_done)); + } + + fn finish_message_if_started(&mut self, events: &mut Vec>) { + if !self.message_started { + return; + } + + let text_done = ResponseStreamEvent::ResponseOutputTextDone(ResponseTextDoneEvent { + sequence_number: self.next_seq(), + item_id: self.message_item_id.clone(), + output_index: self.message_output_index, + content_index: 0, + text: self.accumulated_text.clone(), + logprobs: Some(vec![]), + }); + events.push(make_sse_event(&text_done)); + + let part_done = + ResponseStreamEvent::ResponseContentPartDone(ResponseContentPartDoneEvent { + sequence_number: self.next_seq(), + item_id: self.message_item_id.clone(), + output_index: self.message_output_index, + content_index: 0, + part: OutputContent::OutputText(OutputTextContent { + text: self.accumulated_text.clone(), + annotations: vec![], + logprobs: Some(vec![]), + }), + }); + events.push(make_sse_event(&part_done)); + + let item_done = ResponseStreamEvent::ResponseOutputItemDone(ResponseOutputItemDoneEvent { + sequence_number: self.next_seq(), + output_index: self.message_output_index, + item: OutputItem::Message(OutputMessage { + id: Some(self.message_item_id.clone()), + content: vec![OutputMessageContent::OutputText(OutputTextContent { + text: self.accumulated_text.clone(), + annotations: vec![], + logprobs: Some(vec![]), + })], + role: AssistantRole::Assistant, + status: Some(OutputStatus::Completed), + }), + }); + events.push(make_sse_event(&item_done)); + + self.message_started = false; + self.message_item_id = format!("msg_{}", Uuid::new_v4().simple()); + self.message_output_index = 0; + self.accumulated_text.clear(); + } + fn make_response(&self, status: Status, output: Vec) -> Response { let completed_at = if status == Status::Completed { Some( @@ -220,59 +413,35 @@ impl ResponseStreamConverter { if let Some(content) = content_text && !content.is_empty() { - // Emit output_item.added + content_part.added on first text - if !self.message_started { - self.message_started = true; - self.message_output_index = self.next_output_index; - let output_index = self.message_output_index; - self.next_output_index += 1; - - let item_added = ResponseStreamEvent::ResponseOutputItemAdded( - ResponseOutputItemAddedEvent { - sequence_number: self.next_seq(), - output_index, - item: OutputItem::Message(OutputMessage { - id: Some(self.message_item_id.clone()), - content: vec![], - role: AssistantRole::Assistant, - status: Some(OutputStatus::InProgress), - }), - }, - ); - events.push(make_sse_event(&item_added)); - - let part_added = ResponseStreamEvent::ResponseContentPartAdded( - ResponseContentPartAddedEvent { - sequence_number: self.next_seq(), - item_id: self.message_item_id.clone(), - output_index, - content_index: 0, - part: OutputContent::OutputText(OutputTextContent { - text: String::new(), - annotations: vec![], - logprobs: Some(vec![]), - }), - }, - ); - events.push(make_sse_event(&part_added)); + self.append_raw_text(content); + + // Fallback for models that emit tool calls as raw text instead of + // structured delta.tool_calls chunks. + if delta + .tool_calls + .as_ref() + .is_none_or(|tool_calls| tool_calls.is_empty()) + { + let parsed_calls = parse_tool_call_text(&self.raw_text); + for (name, arguments) in parsed_calls + .iter() + .skip(self.fallback_tool_calls_emitted) + .cloned() + { + self.emit_fallback_tool_call(&mut events, name, arguments); + } + self.fallback_tool_calls_emitted = parsed_calls.len(); } - // Emit text delta - self.accumulated_text.push_str(content); - let text_delta = - ResponseStreamEvent::ResponseOutputTextDelta(ResponseTextDeltaEvent { - sequence_number: self.next_seq(), - item_id: self.message_item_id.clone(), - output_index: self.message_output_index, - content_index: 0, - delta: content.to_string(), - logprobs: Some(vec![]), - }); - events.push(make_sse_event(&text_delta)); + let visible_text = strip_tool_call_text(&self.raw_text).into_owned(); + self.emit_visible_text_delta(&mut events, &visible_text); } // Handle tool call deltas if let Some(tool_calls) = &delta.tool_calls { + if !tool_calls.is_empty() { + self.finish_message_if_started(&mut events); + } for tc in tool_calls { let tc_index = tc.index as usize; @@ -403,46 +572,7 @@ impl ResponseStreamConverter { // Close text message if it was started if self.message_started { - let text_done = ResponseStreamEvent::ResponseOutputTextDone(ResponseTextDoneEvent { - sequence_number: self.next_seq(), - item_id: self.message_item_id.clone(), - output_index: self.message_output_index, - content_index: 0, - text: self.accumulated_text.clone(), - logprobs: Some(vec![]), - }); - events.push(make_sse_event(&text_done)); - - let part_done = - ResponseStreamEvent::ResponseContentPartDone(ResponseContentPartDoneEvent { - sequence_number: self.next_seq(), - item_id: self.message_item_id.clone(), - output_index: self.message_output_index, - content_index: 0, - part: OutputContent::OutputText(OutputTextContent { - text: self.accumulated_text.clone(), - annotations: vec![], - logprobs: Some(vec![]), - }), - }); - events.push(make_sse_event(&part_done)); - - let item_done = - ResponseStreamEvent::ResponseOutputItemDone(ResponseOutputItemDoneEvent { - sequence_number: self.next_seq(), - output_index: self.message_output_index, - item: OutputItem::Message(OutputMessage { - id: Some(self.message_item_id.clone()), - content: vec![OutputMessageContent::OutputText(OutputTextContent { - text: self.accumulated_text.clone(), - annotations: vec![], - logprobs: Some(vec![]), - })], - role: AssistantRole::Assistant, - status: Some(OutputStatus::Completed), - }), - }); - events.push(make_sse_event(&item_done)); + self.finish_message_if_started(&mut events); } // Close any function call items not already done inline From 50e423e21af545742a83537164a3b40495559fa2 Mon Sep 17 00:00:00 2001 From: Ishan Dhanani Date: Wed, 25 Mar 2026 20:00:44 +0000 Subject: [PATCH 2/9] glm47 for codex --- .../src/tool_calling/xml/glm47_parser.rs | 67 +++++++++++++++++-- 1 file changed, 63 insertions(+), 4 deletions(-) diff --git a/lib/parsers/src/tool_calling/xml/glm47_parser.rs b/lib/parsers/src/tool_calling/xml/glm47_parser.rs index 4cdfa5dabe3f..a153fe396b3b 100644 --- a/lib/parsers/src/tool_calling/xml/glm47_parser.rs +++ b/lib/parsers/src/tool_calling/xml/glm47_parser.rs @@ -85,7 +85,7 @@ fn extract_tool_calls( let abs_start = cursor + start_pos; // Add text before tool call to normal parts - normal_parts.push(&text[cursor..abs_start]); + normal_parts.push(text[cursor..abs_start].to_string()); // Find the corresponding end token if let Some(end_pos) = text[abs_start..].find(end_token.as_str()) { @@ -98,19 +98,28 @@ fn extract_tool_calls( Ok(parsed_call) => calls.push(parsed_call), Err(e) => { warn!("Failed to parse GLM-4.7 tool call block: {e}"); - normal_parts.push(block); + if let Some((recovered_text, mut recovered_calls)) = + recover_nested_tool_calls(block, config, tools)? + { + if !recovered_text.is_empty() { + normal_parts.push(recovered_text); + } + calls.append(&mut recovered_calls); + } else { + normal_parts.push(block.to_string()); + } } } cursor = abs_end; } else { // No end token found -> treat the rest as normal text - normal_parts.push(&text[abs_start..]); + normal_parts.push(text[abs_start..].to_string()); break; } } else { // No more tool calls - normal_parts.push(&text[cursor..]); + normal_parts.push(text[cursor..].to_string()); break; } } @@ -119,6 +128,28 @@ fn extract_tool_calls( Ok((normal_text, calls)) } +fn recover_nested_tool_calls( + block: &str, + config: &Glm47ParserConfig, + tools: Option<&[ToolDefinition]>, +) -> anyhow::Result)>> { + let start_token = config.tool_call_start.as_str(); + let nested_start = block[start_token.len()..] + .find(start_token) + .map(|pos| pos + start_token.len()); + let Some(nested_start) = nested_start else { + return Ok(None); + }; + + let nested_block = &block[nested_start..]; + let (normal_text, calls) = extract_tool_calls(nested_block, config, tools)?; + if calls.is_empty() { + return Ok(None); + } + + Ok(Some((normal_text, calls))) +} + /// Decode XML character entities in a string. /// Handles the five predefined XML entities: < > & " ' fn decode_xml_entities(s: &str) -> String { @@ -469,6 +500,34 @@ mod tests { ); } + #[test] + fn test_recovers_nested_valid_tool_call_from_malformed_block() { + let config = get_test_config(); + let tools = vec![ToolDefinition { + name: "spawn_agent".to_string(), + parameters: None, + }]; + + let message = concat!( + "", + "The agent name needs to use only lowercase letters, digits, and underscores. ", + "Let me fix that \"explore_ubuntu_home\" task name.", + "spawn_agent", + "messageinspect repo", + "" + ); + + let (calls, normal_text) = + try_tool_call_parse_glm47(message, &config, Some(&tools)).unwrap(); + + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].function.name, "spawn_agent"); + let args: HashMap = + serde_json::from_str(&calls[0].function.arguments).unwrap(); + assert_eq!(args["message"], "inspect repo"); + assert_eq!(normal_text, Some("".to_string())); + } + #[test] fn test_xml_entity_decoding() { let config = get_test_config(); From 000b07b6d87afdce205820467e5fe30a735235c0 Mon Sep 17 00:00:00 2001 From: Ishan Dhanani Date: Wed, 25 Mar 2026 20:10:46 +0000 Subject: [PATCH 3/9] Fix GLM parser recovery for Codex responses --- .../src/tool_calling/xml/glm47_parser.rs | 61 +++++++++++++++---- 1 file changed, 50 insertions(+), 11 deletions(-) diff --git a/lib/parsers/src/tool_calling/xml/glm47_parser.rs b/lib/parsers/src/tool_calling/xml/glm47_parser.rs index a153fe396b3b..966d5d44e699 100644 --- a/lib/parsers/src/tool_calling/xml/glm47_parser.rs +++ b/lib/parsers/src/tool_calling/xml/glm47_parser.rs @@ -134,20 +134,23 @@ fn recover_nested_tool_calls( tools: Option<&[ToolDefinition]>, ) -> anyhow::Result)>> { let start_token = config.tool_call_start.as_str(); - let nested_start = block[start_token.len()..] - .find(start_token) - .map(|pos| pos + start_token.len()); - let Some(nested_start) = nested_start else { - return Ok(None); - }; + let mut nested_starts = Vec::new(); + let mut search_offset = start_token.len(); + while let Some(relative_start) = block[search_offset..].find(start_token) { + let nested_start = search_offset + relative_start; + nested_starts.push(nested_start); + search_offset = nested_start + start_token.len(); + } - let nested_block = &block[nested_start..]; - let (normal_text, calls) = extract_tool_calls(nested_block, config, tools)?; - if calls.is_empty() { - return Ok(None); + for nested_start in nested_starts.into_iter().rev() { + let nested_block = &block[nested_start..]; + let (normal_text, calls) = extract_tool_calls(nested_block, config, tools)?; + if !calls.is_empty() { + return Ok(Some((normal_text, calls))); + } } - Ok(Some((normal_text, calls))) + Ok(None) } /// Decode XML character entities in a string. @@ -256,6 +259,9 @@ fn parse_tool_call_block( if function_name.is_empty() { anyhow::bail!("Empty function name in tool call"); } + if function_name.contains('<') || function_name.contains('>') { + anyhow::bail!("Malformed function name '{}'", function_name); + } // Parse key-value pairs let mut arguments = HashMap::new(); @@ -528,6 +534,39 @@ mod tests { assert_eq!(normal_text, Some("".to_string())); } + #[test] + fn test_recovers_last_nested_valid_tool_call_from_repeated_prefixes() { + let config = get_test_config(); + let tools = vec![ToolDefinition { + name: "wait_agent".to_string(), + parameters: None, + }]; + + let message = concat!( + "", + "Good, the agent spawned successfully with ID /root/run_ls_simple and nickname Huygens. ", + "Now I need to wait for it to finish its task (running ls) and then close it.", + "Agent spawned. Waiting for it to complete `ls`, then will close.", + "wait", + "Good, the agent spawned successfully with ID /root/run_ls_simple and nickname Huygens. ", + "Now I need to wait for it to finish its task (running ls) and then close it.", + "Agent spawned. Waiting for it to complete `ls`, then will close.", + "wait_agent", + "ids[\"/root/run_ls_simple\"]", + "" + ); + + let (calls, normal_text) = + try_tool_call_parse_glm47(message, &config, Some(&tools)).unwrap(); + + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].function.name, "wait_agent"); + let args: HashMap = + serde_json::from_str(&calls[0].function.arguments).unwrap(); + assert_eq!(args["ids"], serde_json::json!(["/root/run_ls_simple"])); + assert_eq!(normal_text, Some("".to_string())); + } + #[test] fn test_xml_entity_decoding() { let config = get_test_config(); From 68eeede06aa05833e42cdaa5c43d3950fc8230e2 Mon Sep 17 00:00:00 2001 From: Ishan Dhanani Date: Wed, 25 Mar 2026 21:31:55 +0000 Subject: [PATCH 4/9] Improve GLM responses handling for Codex --- .../prompt/template/formatters.rs | 20 +- lib/llm/src/protocols/openai/responses/mod.rs | 207 ++++++++++++++++-- .../openai/responses/stream_converter.rs | 14 ++ .../src/tool_calling/xml/glm47_parser.rs | 159 +++++++++++++- 4 files changed, 363 insertions(+), 37 deletions(-) diff --git a/lib/llm/src/preprocessor/prompt/template/formatters.rs b/lib/llm/src/preprocessor/prompt/template/formatters.rs index 5e483a319e7e..40e092ec2534 100644 --- a/lib/llm/src/preprocessor/prompt/template/formatters.rs +++ b/lib/llm/src/preprocessor/prompt/template/formatters.rs @@ -56,6 +56,10 @@ fn remove_known_non_jinja2_tags(template: &str) -> String { .replace("{% endgeneration %}", "") } +fn rewrite_python_dict_items_calls(template: &str) -> String { + template.replace(".items()", "|items") +} + impl JinjaEnvironment { fn env(self) -> Environment<'static> { self.env @@ -112,7 +116,8 @@ impl HfTokenizerConfigJsonFormatter { supports_add_generation_prompt = Some(true); } // Remove known non-standard tags before validation (they don't affect output) - let template_cleaned = remove_known_non_jinja2_tags(x); + let template_cleaned = + rewrite_python_dict_items_calls(&remove_known_non_jinja2_tags(x)); env.add_template_owned("default", template_cleaned.clone())?; env.add_template_owned("tool_use", template_cleaned)?; } @@ -137,7 +142,8 @@ impl HfTokenizerConfigJsonFormatter { supports_add_generation_prompt = Some(false); } // Remove known non-standard tags before validation (they don't affect output) - let template_cleaned = remove_known_non_jinja2_tags(v); + let template_cleaned = + rewrite_python_dict_items_calls(&remove_known_non_jinja2_tags(v)); env.add_template_owned(k.to_string(), template_cleaned)?; } } @@ -198,4 +204,14 @@ mod tests { let result = remove_known_non_jinja2_tags(template); assert_eq!(result, "Start Part 1 middle Part 2"); } + + #[test] + fn test_rewrite_python_dict_items_calls() { + let template = "{{ args.items() }} {% for k, v in data.items() %}{{ k }}{% endfor %}"; + let result = rewrite_python_dict_items_calls(template); + assert_eq!( + result, + "{{ args|items }} {% for k, v in data|items %}{{ k }}{% endfor %}" + ); + } } diff --git a/lib/llm/src/protocols/openai/responses/mod.rs b/lib/llm/src/protocols/openai/responses/mod.rs index 9355b40aaf86..e9a61a2812be 100644 --- a/lib/llm/src/protocols/openai/responses/mod.rs +++ b/lib/llm/src/protocols/openai/responses/mod.rs @@ -591,32 +591,81 @@ fn parse_tool_call_text(text: &str) -> Vec<(String, String)> { results } +fn extract_think_blocks(text: &str) -> Vec { + if !text.contains("") && !text.contains("") { + return Vec::new(); + } + + let mut extracted = Vec::new(); + let mut remaining = text; + while !remaining.is_empty() { + let next_open = remaining.find(""); + let next_close = remaining.find(""); + + match (next_open, next_close) { + (Some(open), Some(close)) if open < close => { + let content_start = open + "".len(); + let think = remaining[content_start..close].trim(); + if !think.is_empty() { + extracted.push(think.to_string()); + } + remaining = &remaining[close + "".len()..]; + } + (_, Some(close)) => { + let think = remaining[..close].trim(); + if !think.is_empty() { + extracted.push(think.to_string()); + } + remaining = &remaining[close + "".len()..]; + } + (Some(_), None) | (None, None) => break, + } + } + + extracted +} + /// Strip `...` blocks and any `...` blocks from text. /// Returns the original string (no allocation) if no tags are present. fn strip_tool_call_text(text: &str) -> std::borrow::Cow<'_, str> { let has_tool = text.contains(""); - let has_think = text.contains(""); + let has_think = text.contains("") || text.contains(""); if !has_tool && !has_think { return std::borrow::Cow::Borrowed(text); } - fn strip_tag(input: &mut String, open: &str, close: &str) { - while let Some(start) = input.find(open) { - if let Some(end_offset) = input[start..].find(close) { - input.replace_range(start..start + end_offset + close.len(), ""); - } else { - input.truncate(start); - break; + fn strip_tag( + input: &mut String, + open: &str, + close: &str, + strip_prefix_for_dangling_close: bool, + ) { + loop { + let next_open = input.find(open); + let next_close = input.find(close); + + match (next_open, next_close) { + (Some(start), Some(end)) if start < end => { + input.replace_range(start..end + close.len(), ""); + } + (Some(start), _) => { + input.truncate(start); + break; + } + (_, Some(end)) if strip_prefix_for_dangling_close => { + input.replace_range(0..end + close.len(), ""); + } + _ => break, } } } let mut result = text.to_string(); if has_tool { - strip_tag(&mut result, "", ""); + strip_tag(&mut result, "", "", false); } if has_think { - strip_tag(&mut result, "", ""); + strip_tag(&mut result, "", "", true); } std::borrow::Cow::Owned(result) } @@ -719,6 +768,19 @@ pub fn chat_completion_to_response( } // Map reasoning_content to a Reasoning output item + let content_text = match choice.message.content { + Some(dynamo_async_openai::types::ChatCompletionMessageContent::Text(text)) => { + Some(text) + } + Some(dynamo_async_openai::types::ChatCompletionMessageContent::Parts(_)) => { + tracing::warn!( + "Multimodal content in responses API not yet supported, using placeholder" + ); + Some("[multimodal content]".to_string()) + } + None => None, + }; + if let Some(reasoning_text) = choice.message.reasoning_content && !reasoning_text.is_empty() { @@ -731,31 +793,33 @@ pub fn chat_completion_to_response( encrypted_content: None, status: Some(OutputStatus::Completed), })); + } else if let Some(content_text) = content_text.as_ref() { + let think_blocks = extract_think_blocks(content_text); + if !think_blocks.is_empty() { + output.push(OutputItem::Reasoning(ReasoningItem { + id: format!("rs_{}", Uuid::new_v4().simple()), + summary: think_blocks + .into_iter() + .map(|text| SummaryPart::SummaryText(Summary { text })) + .collect(), + content: None, + encrypted_content: None, + status: Some(OutputStatus::Completed), + })); + } } // Handle text content -- also parse blocks from models // that emit tool calls as text (e.g. Qwen3) - let content_text = match choice.message.content { - Some(dynamo_async_openai::types::ChatCompletionMessageContent::Text(text)) => { - Some(text) - } - Some(dynamo_async_openai::types::ChatCompletionMessageContent::Parts(_)) => { - tracing::warn!( - "Multimodal content in responses API not yet supported, using placeholder" - ); - Some("[multimodal content]".to_string()) - } - None => None, - }; if let Some(content_text) = content_text && !content_text.is_empty() { let parsed_calls = parse_tool_call_text(&content_text); + let remaining = strip_tool_call_text(&content_text); if !parsed_calls.is_empty() { for (name, arguments) in parsed_calls { output.push(make_function_call(name, arguments)); } - let remaining = strip_tool_call_text(&content_text); if !remaining.trim().is_empty() { output.push(make_text_message( message_id.clone(), @@ -763,7 +827,10 @@ pub fn chat_completion_to_response( )); } } else { - output.push(make_text_message(message_id.clone(), content_text)); + let visible_text = remaining.into_owned(); + if !visible_text.trim().is_empty() { + output.push(make_text_message(message_id.clone(), visible_text)); + } } } @@ -1321,6 +1388,35 @@ thinking assert!(!stripped.contains("")); } + #[test] + fn test_strip_tool_call_text_with_dangling_think_close_hides_prefix() { + let text = "private reasoningVisible answer."; + let stripped = strip_tool_call_text(text); + assert_eq!(stripped, "Visible answer."); + } + + #[test] + fn test_extract_think_blocks() { + let text = r#" +first + +visible +second"#; + assert_eq!( + extract_think_blocks(text), + vec!["first".to_string(), "second".to_string()] + ); + } + + #[test] + fn test_extract_think_blocks_with_dangling_close() { + let text = "private reasoningVisible answer."; + assert_eq!( + extract_think_blocks(text), + vec!["private reasoning".to_string()] + ); + } + // ── PR1: reasoning / text.format / service_tier pass-through tests ── #[test] @@ -1448,6 +1544,69 @@ thinking assert_eq!(reasoning.effort, Some(ReasoningEffort::High)); } + #[test] + fn test_response_salvages_raw_think_blocks_and_hides_them_from_visible_text() { + let chat_resp = NvCreateChatCompletionResponse { + id: "chatcmpl-think".into(), + choices: vec![dynamo_async_openai::types::ChatChoice { + index: 0, + message: dynamo_async_openai::types::ChatCompletionResponseMessage { + content: Some( + dynamo_async_openai::types::ChatCompletionMessageContent::Text( + "private chain of thoughtPublic answer.".to_string(), + ), + ), + refusal: None, + tool_calls: None, + role: dynamo_async_openai::types::Role::Assistant, + function_call: None, + audio: None, + reasoning_content: None, + }, + finish_reason: None, + stop_reason: None, + logprobs: None, + }], + created: 0, + model: "test-model".into(), + service_tier: None, + system_fingerprint: None, + object: "chat.completion".to_string(), + usage: None, + nvext: None, + }; + + let wrapped = chat_completion_to_response(chat_resp, &ResponseParams::default()).unwrap(); + assert_eq!(wrapped.inner.output.len(), 2); + + match &wrapped.inner.output[0] { + OutputItem::Reasoning(reasoning) => { + assert_eq!(reasoning.summary.len(), 1); + match &reasoning.summary[0] { + SummaryPart::SummaryText(summary) => { + assert_eq!(summary.text, "private chain of thought"); + } + _ => panic!("expected reasoning summary text"), + } + } + other => panic!("expected reasoning item, got {other:?}"), + } + + match &wrapped.inner.output[1] { + OutputItem::Message(message) => { + assert_eq!(message.content.len(), 1); + match &message.content[0] { + OutputMessageContent::OutputText(text) => { + assert_eq!(text.text, "Public answer."); + assert!(!text.text.contains("")); + } + _ => panic!("expected output text"), + } + } + other => panic!("expected output message, got {other:?}"), + } + } + #[test] fn test_response_echoes_text_format() { use dynamo_async_openai::types::responses::{ diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index d81b311dbc81..0ce4153d94cb 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -1038,4 +1038,18 @@ mod tests { "output_item.done inline after text: {tool_types:?}" ); } + + #[test] + fn test_dangling_think_close_does_not_leak_hidden_prefix() { + let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); + let _ = conv.emit_start_events(); + + let events = conv.process_chunk(&text_chunk("private reasoningVisible answer.")); + let types = event_types(&events); + assert!( + types.contains(&"response.output_text.delta".to_string()), + "visible text delta should still be emitted: {types:?}" + ); + assert_eq!(conv.accumulated_text, "Visible answer."); + } } diff --git a/lib/parsers/src/tool_calling/xml/glm47_parser.rs b/lib/parsers/src/tool_calling/xml/glm47_parser.rs index 966d5d44e699..a67e02b3f293 100644 --- a/lib/parsers/src/tool_calling/xml/glm47_parser.rs +++ b/lib/parsers/src/tool_calling/xml/glm47_parser.rs @@ -97,7 +97,6 @@ fn extract_tool_calls( match parse_tool_call_block(block, config, tools) { Ok(parsed_call) => calls.push(parsed_call), Err(e) => { - warn!("Failed to parse GLM-4.7 tool call block: {e}"); if let Some((recovered_text, mut recovered_calls)) = recover_nested_tool_calls(block, config, tools)? { @@ -106,6 +105,7 @@ fn extract_tool_calls( } calls.append(&mut recovered_calls); } else { + warn!("Failed to parse GLM-4.7 tool call block: {e}"); normal_parts.push(block.to_string()); } } @@ -134,15 +134,33 @@ fn recover_nested_tool_calls( tools: Option<&[ToolDefinition]>, ) -> anyhow::Result)>> { let start_token = config.tool_call_start.as_str(); - let mut nested_starts = Vec::new(); - let mut search_offset = start_token.len(); - while let Some(relative_start) = block[search_offset..].find(start_token) { - let nested_start = search_offset + relative_start; - nested_starts.push(nested_start); - search_offset = nested_start + start_token.len(); + let mut candidate_starts = Vec::new(); + + if let Some(tools_list) = tools { + for tool in tools_list { + let exact_start = format!("{start_token}{}", tool.name); + let mut search_offset = start_token.len(); + while let Some(relative_start) = block[search_offset..].find(&exact_start) { + let nested_start = search_offset + relative_start; + candidate_starts.push(nested_start); + search_offset = nested_start + 1; + } + } } - for nested_start in nested_starts.into_iter().rev() { + if candidate_starts.is_empty() { + let mut search_offset = start_token.len(); + while let Some(relative_start) = block[search_offset..].find(start_token) { + let nested_start = search_offset + relative_start; + candidate_starts.push(nested_start); + search_offset = nested_start + start_token.len(); + } + } + + candidate_starts.sort_unstable(); + candidate_starts.dedup(); + + for nested_start in candidate_starts.into_iter().rev() { let nested_block = &block[nested_start..]; let (normal_text, calls) = extract_tool_calls(nested_block, config, tools)?; if !calls.is_empty() { @@ -231,6 +249,25 @@ fn get_param_schema_type<'a>( param.get("type")?.as_str() } +fn recover_malformed_function_name_prefix<'a>( + raw_function_name: &'a str, + tools: Option<&'a [ToolDefinition]>, +) -> Option<&'a str> { + let tools = tools?; + + tools + .iter() + .map(|tool| tool.name.as_str()) + .filter(|tool_name| { + raw_function_name.starts_with(tool_name) + && raw_function_name[tool_name.len()..] + .chars() + .next() + .is_none_or(|ch| !matches!(ch, 'a'..='z' | '0'..='9' | '_')) + }) + .max_by_key(|tool_name| tool_name.len()) +} + /// Parse a single GLM-4.7 tool call block /// Format: function_namekey1value1... fn parse_tool_call_block( @@ -249,18 +286,22 @@ fn parse_tool_call_block( // Extract function name (everything before first or end) let arg_key_start = &config.arg_key_start; - let function_name = if let Some(pos) = content.find(arg_key_start.as_str()) { + let raw_function_name = if let Some(pos) = content.find(arg_key_start.as_str()) { content[..pos].trim().to_string() } else { // No arguments, just function name content.trim().to_string() }; - if function_name.is_empty() { + if raw_function_name.is_empty() { anyhow::bail!("Empty function name in tool call"); } + let function_name = recover_malformed_function_name_prefix(&raw_function_name, tools) + .unwrap_or(raw_function_name.as_str()) + .to_string(); + if function_name.contains('<') || function_name.contains('>') { - anyhow::bail!("Malformed function name '{}'", function_name); + anyhow::bail!("Malformed function name '{function_name}'"); } // Parse key-value pairs @@ -567,6 +608,102 @@ mod tests { assert_eq!(normal_text, Some("".to_string())); } + #[test] + fn test_recovers_valid_tool_name_prefix_from_malformed_function_name() { + let config = get_test_config(); + let tools = vec![ToolDefinition { + name: "wait_agent".to_string(), + parameters: None, + }]; + + let message = concat!( + "", + "wait_agent", + "The agent has been spawned successfully with id \"019d26c6-bcc7-7152-be59-dc68332552f7\". ", + "Now I should wait for it to complete and then close it.", + "Agent spindle started with ID `019d26c6-bcc7-7152-be59-dc68332552f7`. Waiting for completion then closing it.", + "ids[\"019d26c6-bcc7-7152-be59-dc68332552f7\"]", + "" + ); + + let (calls, normal_text) = + try_tool_call_parse_glm47(message, &config, Some(&tools)).unwrap(); + + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].function.name, "wait_agent"); + let args: HashMap = + serde_json::from_str(&calls[0].function.arguments).unwrap(); + assert_eq!( + args["ids"], + serde_json::json!(["019d26c6-bcc7-7152-be59-dc68332552f7"]) + ); + assert_eq!(normal_text, Some("".to_string())); + } + + #[test] + fn test_recovers_nested_tool_call_with_malformed_exact_name_start() { + let config = get_test_config(); + let tools = vec![ToolDefinition { + name: "wait_agent".to_string(), + parameters: None, + }]; + + let message = concat!( + "", + "Great, the agent has been spawned. Now I need to wait for it to complete before closing it.", + "wait_agent", + "Great, the agent has been spawned. Now I need to wait for it to complete before closing it.", + "ids[\"019d26cb-3125-7e03-a07c-363ebffae56b\"]", + "" + ); + + let (calls, normal_text) = + try_tool_call_parse_glm47(message, &config, Some(&tools)).unwrap(); + + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].function.name, "wait_agent"); + let args: HashMap = + serde_json::from_str(&calls[0].function.arguments).unwrap(); + assert_eq!( + args["ids"], + serde_json::json!(["019d26cb-3125-7e03-a07c-363ebffae56b"]) + ); + assert_eq!(normal_text, Some("".to_string())); + } + + #[test] + fn test_recovers_exact_tool_name_start_from_repeated_spawn_agent_prefixes() { + let config = get_test_config(); + let tools = vec![ToolDefinition { + name: "spawn_agent".to_string(), + parameters: None, + }]; + + let message = concat!( + "", + "I got an error - the agent_name must be lowercase with only letters, digits, and underscores. ", + "Let me fix that.", + "spawn", + "I got an error - the agent_name must be lowercase with only letters, digits, and underscores. ", + "Let me fix that.", + "spawn_agent", + "agent_namels_executor", + "messageRun ls and report output", + "" + ); + + let (calls, normal_text) = + try_tool_call_parse_glm47(message, &config, Some(&tools)).unwrap(); + + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].function.name, "spawn_agent"); + let args: HashMap = + serde_json::from_str(&calls[0].function.arguments).unwrap(); + assert_eq!(args["agent_name"], "ls_executor"); + assert_eq!(args["message"], "Run ls and report output"); + assert_eq!(normal_text, Some("".to_string())); + } + #[test] fn test_xml_entity_decoding() { let config = get_test_config(); From f42eff8c68819273004ab9822520c39dd31f3660 Mon Sep 17 00:00:00 2001 From: Ishan Dhanani Date: Fri, 27 Mar 2026 13:18:36 +0000 Subject: [PATCH 5/9] test: clean up responses reasoning coverage --- lib/llm/src/protocols/openai/responses/mod.rs | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/lib/llm/src/protocols/openai/responses/mod.rs b/lib/llm/src/protocols/openai/responses/mod.rs index e9a61a2812be..d0ff29ddfe0e 100644 --- a/lib/llm/src/protocols/openai/responses/mod.rs +++ b/lib/llm/src/protocols/openai/responses/mod.rs @@ -1546,6 +1546,7 @@ visible #[test] fn test_response_salvages_raw_think_blocks_and_hides_them_from_visible_text() { + #[allow(deprecated)] let chat_resp = NvCreateChatCompletionResponse { id: "chatcmpl-think".into(), choices: vec![dynamo_async_openai::types::ChatChoice { @@ -1582,12 +1583,8 @@ visible match &wrapped.inner.output[0] { OutputItem::Reasoning(reasoning) => { assert_eq!(reasoning.summary.len(), 1); - match &reasoning.summary[0] { - SummaryPart::SummaryText(summary) => { - assert_eq!(summary.text, "private chain of thought"); - } - _ => panic!("expected reasoning summary text"), - } + let SummaryPart::SummaryText(summary) = &reasoning.summary[0]; + assert_eq!(summary.text, "private chain of thought"); } other => panic!("expected reasoning item, got {other:?}"), } From 02a21acfe7864221e482891d8b31ba957d6cca2b Mon Sep 17 00:00:00 2001 From: Ishan Dhanani Date: Fri, 27 Mar 2026 13:41:52 +0000 Subject: [PATCH 6/9] fix: dedupe mixed responses tool call streams --- .../openai/responses/stream_converter.rs | 235 +++++++++++++----- 1 file changed, 176 insertions(+), 59 deletions(-) diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index 0ce4153d94cb..6e7f59f77619 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -47,6 +47,7 @@ pub struct ResponseStreamConverter { // Function call tracking function_call_items: Vec, fallback_tool_calls_emitted: usize, + structured_tool_calls_seen: bool, // Output index counter next_output_index: u32, // Usage stats from the backend's final chunk @@ -85,6 +86,7 @@ impl ResponseStreamConverter { accumulated_text: String::new(), function_call_items: Vec::new(), fallback_tool_calls_emitted: 0, + structured_tool_calls_seen: false, next_output_index: 0, usage: None, } @@ -417,7 +419,8 @@ impl ResponseStreamConverter { // Fallback for models that emit tool calls as raw text instead of // structured delta.tool_calls chunks. - if delta + if !self.structured_tool_calls_seen + && delta .tool_calls .as_ref() .is_none_or(|tool_calls| tool_calls.is_empty()) @@ -440,11 +443,24 @@ impl ResponseStreamConverter { // Handle tool call deltas if let Some(tool_calls) = &delta.tool_calls { if !tool_calls.is_empty() { + self.structured_tool_calls_seen = true; + self.fallback_tool_calls_emitted = self + .fallback_tool_calls_emitted + .max(parse_tool_call_text(&self.raw_text).len()); self.finish_message_if_started(&mut events); } for tc in tool_calls { let tc_index = tc.index as usize; + if tc_index < self.fallback_tool_calls_emitted + && self + .function_call_items + .get(tc_index) + .is_some_and(|fc| fc.started && fc.done) + { + continue; + } + // Start a new function call if we haven't seen this index while self.function_call_items.len() <= tc_index { let output_index = self.next_output_index; @@ -784,10 +800,7 @@ fn get_event_type(event: &ResponseStreamEvent) -> &'static str { #[cfg(test)] mod tests { use super::*; - use dynamo_async_openai::types::{ - ChatChoiceStream, ChatCompletionMessageContent, ChatCompletionMessageToolCallChunk, - ChatCompletionStreamResponseDelta, ChatCompletionToolType, FunctionCallStream, - }; + use serde_json::json; fn default_params() -> ResponseParams { ResponseParams { @@ -813,67 +826,65 @@ mod tests { name: Option<&str>, args: Option<&str>, ) -> NvCreateChatCompletionStreamResponse { - #[allow(deprecated)] - NvCreateChatCompletionStreamResponse { - id: "chat-1".into(), - choices: vec![ChatChoiceStream { - index: 0, - delta: ChatCompletionStreamResponseDelta { - content: None, - function_call: None, - tool_calls: Some(vec![ChatCompletionMessageToolCallChunk { - index: tc_index, - id: id.map(String::from), - r#type: Some(ChatCompletionToolType::Function), - function: Some(FunctionCallStream { - name: name.map(String::from), - arguments: args.map(String::from), - }), - }]), - role: None, - refusal: None, - reasoning_content: None, + serde_json::from_value(json!({ + "id": "chat-1", + "choices": [{ + "index": 0, + "delta": { + "content": null, + "tool_calls": [{ + "index": tc_index, + "id": id, + "type": "function", + "function": { + "name": name, + "arguments": args, + } + }], + "role": null, + "refusal": null, + "reasoning_content": null }, - finish_reason: None, - stop_reason: None, - logprobs: None, + "finish_reason": null, + "stop_reason": null, + "logprobs": null }], - created: 0, - model: "test".into(), - service_tier: None, - system_fingerprint: None, - object: "chat.completion.chunk".into(), - usage: None, - nvext: None, - } + "created": 0, + "model": "test", + "service_tier": null, + "system_fingerprint": null, + "object": "chat.completion.chunk", + "usage": null, + "nvext": null + })) + .expect("tool call test fixture should deserialize") } fn text_chunk(text: &str) -> NvCreateChatCompletionStreamResponse { - #[allow(deprecated)] - NvCreateChatCompletionStreamResponse { - id: "chat-1".into(), - choices: vec![ChatChoiceStream { - index: 0, - delta: ChatCompletionStreamResponseDelta { - content: Some(ChatCompletionMessageContent::Text(text.into())), - function_call: None, - tool_calls: None, - role: None, - refusal: None, - reasoning_content: None, + serde_json::from_value(json!({ + "id": "chat-1", + "choices": [{ + "index": 0, + "delta": { + "content": text, + "tool_calls": null, + "role": null, + "refusal": null, + "reasoning_content": null }, - finish_reason: None, - stop_reason: None, - logprobs: None, + "finish_reason": null, + "stop_reason": null, + "logprobs": null }], - created: 0, - model: "test".into(), - service_tier: None, - system_fingerprint: None, - object: "chat.completion.chunk".into(), - usage: None, - nvext: None, - } + "created": 0, + "model": "test", + "service_tier": null, + "system_fingerprint": null, + "object": "chat.completion.chunk", + "usage": null, + "nvext": null + })) + .expect("text test fixture should deserialize") } /// Extract the SSE event type from a Result. @@ -1052,4 +1063,110 @@ mod tests { ); assert_eq!(conv.accumulated_text, "Visible answer."); } + + #[test] + fn test_structured_tool_call_does_not_duplicate_raw_tool_call_text_same_chunk() { + let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); + let _ = conv.emit_start_events(); + + let events = conv.process_chunk( + &serde_json::from_value(json!({ + "id": "test", + "choices": [{ + "index": 0, + "delta": { + "content": "{\"name\":\"spawn_agent\",\"arguments\":{\"task\":\"ls\"}}", + "tool_calls": [{ + "index": 0, + "id": "call-1", + "type": "function", + "function": { + "name": "spawn_agent", + "arguments": "{\"task\":\"ls\"}" + } + }], + "role": null, + "refusal": null, + "reasoning_content": null + }, + "finish_reason": null, + "stop_reason": null, + "logprobs": null + }], + "created": 0, + "model": "test", + "service_tier": null, + "system_fingerprint": null, + "object": "chat.completion.chunk", + "usage": null, + "nvext": null + })) + .expect("combined content/tool-calls fixture should deserialize"), + ); + + let types = event_types(&events); + assert_eq!( + types + .iter() + .filter(|t| *t == "response.output_item.added") + .count(), + 1, + "should emit exactly one function-call item: {types:?}" + ); + assert_eq!( + types + .iter() + .filter(|t| *t == "response.function_call_arguments.done") + .count(), + 1, + "should emit exactly one function-call args done: {types:?}" + ); + assert_eq!( + types + .iter() + .filter(|t| *t == "response.output_item.done") + .count(), + 1, + "should emit exactly one function-call item done: {types:?}" + ); + } + + #[test] + fn test_structured_tool_call_does_not_duplicate_prior_fallback_tool_call() { + let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); + let _ = conv.emit_start_events(); + + let fallback_events = conv.process_chunk(&text_chunk( + "{\"name\":\"spawn_agent\",\"arguments\":{\"task\":\"ls\"}}", + )); + let fallback_types = event_types(&fallback_events); + assert_eq!( + fallback_types + .iter() + .filter(|t| *t == "response.output_item.done") + .count(), + 1, + "fallback path should emit one completed function call: {fallback_types:?}" + ); + + let structured_events = conv.process_chunk(&tool_call_chunk( + 0, + Some("call-1"), + Some("spawn_agent"), + Some("{\"task\":\"ls\"}"), + )); + let structured_types = event_types(&structured_events); + assert!( + !structured_types.contains(&"response.output_item.added".to_string()), + "structured duplicate should be ignored after fallback: {structured_types:?}" + ); + assert!( + !structured_types.contains(&"response.function_call_arguments.done".to_string()), + "structured duplicate args should be ignored after fallback: {structured_types:?}" + ); + assert!( + !structured_types.contains(&"response.output_item.done".to_string()), + "structured duplicate completion should be ignored after fallback: {structured_types:?}" + ); + } } From f2b5d1ab9f79a72e2996e047792dc124dbcc3105 Mon Sep 17 00:00:00 2001 From: Ishan Dhanani Date: Fri, 27 Mar 2026 13:55:07 +0000 Subject: [PATCH 7/9] style: format responses stream converter --- lib/llm/src/protocols/openai/responses/stream_converter.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index 6e7f59f77619..a4ed427b4f9f 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -421,9 +421,9 @@ impl ResponseStreamConverter { // structured delta.tool_calls chunks. if !self.structured_tool_calls_seen && delta - .tool_calls - .as_ref() - .is_none_or(|tool_calls| tool_calls.is_empty()) + .tool_calls + .as_ref() + .is_none_or(|tool_calls| tool_calls.is_empty()) { let parsed_calls = parse_tool_call_text(&self.raw_text); for (name, arguments) in parsed_calls From 19e39c6a779182b56f8f06404f3c11bed145915f Mon Sep 17 00:00:00 2001 From: Matej Kosec Date: Sat, 28 Mar 2026 20:24:26 -0700 Subject: [PATCH 8/9] fix: preserve text output in response.completed and dedupe non-streaming tool calls Two bugs addressed: 1. Streaming: finish_message_if_started() clears message_started and accumulated_text, so the subsequent guard in emit_end_events() that builds Response.output always evaluated false. Text-only streams got an empty output array in the response.completed event. Fix: capture message state before calling finish. 2. Non-streaming: chat_completion_to_response() emitted structured tool_calls first, then unconditionally re-parsed content for raw blocks, producing duplicates when both were present. Fix: skip raw parsing when structured tool_calls already exist. Signed-off-by: Matej Kosec --- lib/llm/src/protocols/openai/responses/mod.rs | 14 ++++++- .../openai/responses/stream_converter.rs | 42 +++++++++++++++++-- 2 files changed, 50 insertions(+), 6 deletions(-) diff --git a/lib/llm/src/protocols/openai/responses/mod.rs b/lib/llm/src/protocols/openai/responses/mod.rs index d0ff29ddfe0e..b656c8d550ba 100644 --- a/lib/llm/src/protocols/openai/responses/mod.rs +++ b/lib/llm/src/protocols/openai/responses/mod.rs @@ -755,6 +755,11 @@ pub fn chat_completion_to_response( if let Some(choice) = choice { // Handle structured tool calls + let has_structured_tool_calls = choice + .message + .tool_calls + .as_ref() + .is_some_and(|tc| !tc.is_empty()); if let Some(tool_calls) = choice.message.tool_calls { for tc in &tool_calls { output.push(OutputItem::FunctionCall(FunctionToolCall { @@ -810,11 +815,16 @@ pub fn chat_completion_to_response( } // Handle text content -- also parse blocks from models - // that emit tool calls as text (e.g. Qwen3) + // that emit tool calls as text (e.g. Qwen3). + // Skip raw parsing when structured tool_calls already exist to avoid duplicates. if let Some(content_text) = content_text && !content_text.is_empty() { - let parsed_calls = parse_tool_call_text(&content_text); + let parsed_calls = if has_structured_tool_calls { + Vec::new() + } else { + parse_tool_call_text(&content_text) + }; let remaining = strip_tool_call_text(&content_text); if !parsed_calls.is_empty() { for (name, arguments) in parsed_calls { diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index a4ed427b4f9f..9df34120ae1b 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -586,6 +586,13 @@ impl ResponseStreamConverter { pub fn emit_end_events(&mut self) -> Vec> { let mut events = Vec::new(); + // Capture message state before finishing (finish clears the fields) + let pending_message = if self.message_started { + Some((self.message_item_id.clone(), self.accumulated_text.clone())) + } else { + None + }; + // Close text message if it was started if self.message_started { self.finish_message_if_started(&mut events); @@ -633,13 +640,13 @@ impl ResponseStreamConverter { events.push(make_sse_event(&item_done)); } - // Build the final output vector from accumulated state + // Build the final output vector from captured state let mut output = Vec::new(); - if self.message_started { + if let Some((msg_id, text)) = pending_message { output.push(OutputItem::Message(OutputMessage { - id: Some(self.message_item_id.clone()), + id: Some(msg_id), content: vec![OutputMessageContent::OutputText(OutputTextContent { - text: self.accumulated_text.clone(), + text, annotations: vec![], logprobs: Some(vec![]), })], @@ -1050,6 +1057,33 @@ mod tests { ); } + #[test] + fn test_text_only_response_completed_includes_output() { + let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); + let _ = conv.emit_start_events(); + let _ = conv.process_chunk(&text_chunk("Hello world")); + let end_events = conv.emit_end_events(); + let end_types = event_types(&end_events); + + assert!( + end_types.contains(&"response.completed".to_string()), + "should contain response.completed: {end_types:?}" + ); + + // Verify the response.completed event includes the text in output + for event in &end_events { + let serialized = format!("{:?}", event.as_ref().unwrap()); + if serialized.contains("response.completed") { + assert!( + serialized.contains("Hello world"), + "response.completed output should contain the accumulated text" + ); + return; + } + } + panic!("response.completed event not found in end events"); + } + #[test] fn test_dangling_think_close_does_not_leak_hidden_prefix() { let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); From 533f813b52851082050fe766bf19e1285fb25788 Mon Sep 17 00:00:00 2001 From: Matej Kosec Date: Sat, 28 Mar 2026 20:44:35 -0700 Subject: [PATCH 9/9] fix: handle retroactive think-tag stripping across streaming chunks Two bugs addressed: 1. strip_tag arm ordering: when appeared before (dangling close followed by a matched pair), the (Some(start), _) arm fired instead of the dangling-close arm, leaking the tag and truncating everything after the open. Fix: reorder match arms so dangling-close check runs before open-without-close. 2. emit_visible_text_delta prefix assumption: when retroactive stripping (e.g. completing across chunks) invalidated already-emitted text, strip_prefix returned None and accumulated_text was never updated, permanently blocking all future deltas. Fix: fall back to longest common prefix computation so subsequent text is still emitted. Signed-off-by: Matej Kosec --- lib/llm/src/protocols/openai/responses/mod.rs | 17 ++++-- .../openai/responses/stream_converter.rs | 53 ++++++++++++++++++- 2 files changed, 66 insertions(+), 4 deletions(-) diff --git a/lib/llm/src/protocols/openai/responses/mod.rs b/lib/llm/src/protocols/openai/responses/mod.rs index b656c8d550ba..7395e8c9ee43 100644 --- a/lib/llm/src/protocols/openai/responses/mod.rs +++ b/lib/llm/src/protocols/openai/responses/mod.rs @@ -646,15 +646,19 @@ fn strip_tool_call_text(text: &str) -> std::borrow::Cow<'_, str> { match (next_open, next_close) { (Some(start), Some(end)) if start < end => { + // Matched pair: ... input.replace_range(start..end + close.len(), ""); } + (_, Some(end)) if strip_prefix_for_dangling_close => { + // Dangling close before any open (or open comes after close): + // strip everything up to and including the close tag. + input.replace_range(0..end + close.len(), ""); + } (Some(start), _) => { + // Open without matching close: truncate at open position. input.truncate(start); break; } - (_, Some(end)) if strip_prefix_for_dangling_close => { - input.replace_range(0..end + close.len(), ""); - } _ => break, } } @@ -1405,6 +1409,13 @@ thinking assert_eq!(stripped, "Visible answer."); } + #[test] + fn test_strip_tool_call_text_dangling_close_then_matched_pair() { + let text = "thinkingvisiblehiddenanswer"; + let stripped = strip_tool_call_text(text); + assert_eq!(stripped, "visibleanswer"); + } + #[test] fn test_extract_think_blocks() { let text = r#" diff --git a/lib/llm/src/protocols/openai/responses/stream_converter.rs b/lib/llm/src/protocols/openai/responses/stream_converter.rs index 9df34120ae1b..e7334e08b310 100644 --- a/lib/llm/src/protocols/openai/responses/stream_converter.rs +++ b/lib/llm/src/protocols/openai/responses/stream_converter.rs @@ -153,10 +153,29 @@ impl ResponseStreamConverter { events: &mut Vec>, visible_text: &str, ) { - let Some(delta) = visible_text.strip_prefix(&self.accumulated_text) else { + if visible_text == self.accumulated_text { return; + } + + // Normally visible_text extends accumulated_text, but retroactive stripping + // (e.g. a dangling completing across chunks) can invalidate the + // prefix assumption. Fall back to longest common prefix so we don't + // permanently lose all subsequent text. + let delta = if let Some(suffix) = visible_text.strip_prefix(&self.accumulated_text) { + suffix + } else { + let common_byte_len: usize = self + .accumulated_text + .chars() + .zip(visible_text.chars()) + .take_while(|(a, b)| a == b) + .map(|(c, _)| c.len_utf8()) + .sum(); + &visible_text[common_byte_len..] }; + if delta.is_empty() { + self.accumulated_text = visible_text.to_string(); return; } @@ -1098,6 +1117,38 @@ mod tests { assert_eq!(conv.accumulated_text, "Visible answer."); } + #[test] + fn test_multi_chunk_dangling_think_close_does_not_drop_subsequent_text() { + let mut conv = ResponseStreamConverter::new("test-model".into(), default_params()); + let _ = conv.emit_start_events(); + + // Chunk 1: thinking text arrives without tags — appears visible initially + let events1 = conv.process_chunk(&text_chunk("private reasoning")); + let types1 = event_types(&events1); + assert!( + types1.contains(&"response.output_text.delta".to_string()), + "chunk 1 should emit visible delta: {types1:?}" + ); + + // Chunk 2: completes — retroactively hides chunk 1's text + let events2 = conv.process_chunk(&text_chunk("Visible answer.")); + let types2 = event_types(&events2); + assert!( + types2.contains(&"response.output_text.delta".to_string()), + "chunk 2 must still emit visible text after : {types2:?}" + ); + assert_eq!(conv.accumulated_text, "Visible answer."); + + // Chunk 3: subsequent text must not be permanently dropped + let events3 = conv.process_chunk(&text_chunk(" More text.")); + let types3 = event_types(&events3); + assert!( + types3.contains(&"response.output_text.delta".to_string()), + "chunk 3 must not be dropped: {types3:?}" + ); + assert_eq!(conv.accumulated_text, "Visible answer. More text."); + } + #[test] fn test_structured_tool_call_does_not_duplicate_raw_tool_call_text_same_chunk() { let mut conv = ResponseStreamConverter::new("test-model".into(), default_params());