From 39cdedb7a40d9157eac4c5fbbbb940fd33b16e11 Mon Sep 17 00:00:00 2001 From: Zihua Wu <13583761+lucifer1004@users.noreply.github.com> Date: Thu, 20 Aug 2026 02:51:19 -0700 Subject: [PATCH 1/3] feat(tool-parser): flush DSML terminal tool calls and held-back text at stream end Support DeepSeek V4 reasoning plus automatic tool calls: treat any DSML sentinel as structural, validate invoke arguments before completing a call, and add ToolParser::finish_incremental/completed_tool_call_count so the gRPC streaming router can flush finalized calls, unstreamed args, and retained normal text at the terminal boundary. Signed-off-by: Zihua Wu <13583761+lucifer1004@users.noreply.github.com> --- .../tool_parser/src/parsers/deepseek_dsml.rs | 137 +++++++++++++++--- crates/tool_parser/src/traits.rs | 14 ++ .../tests/tool_parser_deepseek_dsml.rs | 94 ++++++++++-- .../src/routers/grpc/regular/streaming.rs | 77 ++++++---- 4 files changed, 263 insertions(+), 59 deletions(-) diff --git a/crates/tool_parser/src/parsers/deepseek_dsml.rs b/crates/tool_parser/src/parsers/deepseek_dsml.rs index d1fe28668c..d46378c20a 100644 --- a/crates/tool_parser/src/parsers/deepseek_dsml.rs +++ b/crates/tool_parser/src/parsers/deepseek_dsml.rs @@ -4,7 +4,7 @@ use regex::Regex; use serde_json::Value; use crate::{ - errors::{ParserError, ParserResult}, + errors::ParserResult, parsers::helpers, traits::ToolParser, types::{FunctionCall, StreamingParseResult, ToolCall, ToolCallItem}, @@ -59,11 +59,15 @@ pub struct DeepSeekDsmlParser { current_tool_name_sent: bool, /// Tracks raw JSON string content streamed to client for each tool's arguments streamed_args_for_tool: Vec, + /// Calls whose invoke and argument object have both closed successfully. + completed_tool_call_count: usize, } /// Full DSML closing tags for suffix-based stripping during streaming. const DSML_PARAMETER_END_TAG: &str = ""; const DSML_INVOKE_END_TAG: &str = ""; +const DSML_OPEN_SENTINEL: &str = "<｜DSML｜"; +const DSML_CLOSE_SENTINEL: &str = " ToolCall { + fn arguments_are_valid_object(content: &str, arguments: &str) -> bool { + let candidate = if content.trim().starts_with('{') { + content.trim() + } else { + arguments + }; + matches!(serde_json::from_str(candidate), Ok(Value::Object(_))) + } + + /// Parse a single complete invoke block into a ToolCall. + fn parse_invoke(&self, name: &str, content: &str) -> Option { let arguments = self.parse_parameters_from_dsml(content, false); + if !Self::arguments_are_valid_object(content, &arguments) { + return None; + } - ToolCall { + Some(ToolCall { function: FunctionCall { name: name.trim().to_string(), arguments, }, - } + }) } } @@ -257,31 +274,54 @@ impl ToolParser for DeepSeekDsmlParser { return Ok((text.to_string(), vec![])); } - let idx = text - .find(self.block_open.as_str()) - .ok_or_else(|| ParserError::ParsingFailed("DSML marker not found".to_string()))?; + let Some(idx) = text.find(self.block_open.as_str()) else { + let marker_start = [DSML_OPEN_SENTINEL, DSML_CLOSE_SENTINEL] + .into_iter() + .filter_map(|marker| text.find(marker)) + .min(); + let Some(marker_start) = marker_start else { + return Ok((text.to_string(), Vec::new())); + }; + return Ok((text[..marker_start].trim_end().to_string(), Vec::new())); + }; let normal_text = text[..idx].trim_end().to_string(); let mut tools = Vec::new(); + let mut last_block_end = None; for fc_cap in self.tool_call_complete_regex.captures_iter(text) { + last_block_end = fc_cap.get(0).map(|outer| outer.end()); let fc_content = fc_cap.get(1).map_or("", |m| m.as_str()); for inv_cap in self.invoke_complete_regex.captures_iter(fc_content) { let func_name = inv_cap.get(1).map_or("", |m| m.as_str()); let invoke_content = inv_cap.get(2).map_or("", |m| m.as_str()); - tools.push(self.parse_invoke(func_name, invoke_content)); + if let Some(tool) = self.parse_invoke(func_name, invoke_content) { + tools.push(tool); + } } } - if tools.is_empty() { - return Ok((normal_text, vec![])); + let mut normal_text = normal_text; + if let Some(end) = last_block_end { + normal_text.push_str(&text[end..].replace(EOS_TOKEN, "")); } Ok((normal_text, tools)) } + async fn parse_complete_with_tools( + &self, + output: &str, + tools: &[Tool], + ) -> ParserResult<(String, Vec)> { + let (normal_text, mut calls) = self.parse_complete(output).await?; + let tool_indices = helpers::get_tool_indices(tools); + calls.retain(|call| tool_indices.contains_key(call.function.name.as_str())); + Ok((normal_text, calls)) + } + async fn parse_incremental( &mut self, chunk: &str, @@ -302,13 +342,19 @@ impl ToolParser for DeepSeekDsmlParser { // passthrough path and lose the sentinel, turning every subsequent // chunk into plain text. (See regression test // `test_deepseek_dsml_v4_streaming_bpe_chunked_opener`.) - let has_dsml = current_text.contains("<｜DSML｜"); - let has_partial_prefix = current_text.ends_with('<') - || current_text.ends_with("<｜") - || current_text.ends_with(" = Vec::new(); @@ -379,12 +431,25 @@ impl ToolParser for DeepSeekDsmlParser { &self.prev_tool_call_arr, ); return Ok(StreamingParseResult { - normal_text: String::new(), + normal_text, calls: all_calls, }); } } + let current_args = self.parse_parameters_from_dsml(&invoke_content, !is_complete); + if is_complete && !Self::arguments_are_valid_object(&invoke_content, ¤t_args) { + tracing::debug!("Invalid arguments for tool '{}' - skipping", func_name); + if let Some(end) = match_end { + self.buffer = self.buffer[end..].to_string(); + } + if self.current_tool_name_sent { + self.current_tool_id += 1; + } + self.current_tool_name_sent = false; + continue; + } + // Initialize state on first tool if self.current_tool_id == -1 { self.current_tool_id = 0; @@ -418,8 +483,6 @@ impl ToolParser for DeepSeekDsmlParser { }); } - // Parse current arguments (partial or complete) - let current_args = self.parse_parameters_from_dsml(&invoke_content, !is_complete); let tool_id = self.current_tool_id as usize; // Compute diff against what we've already sent @@ -492,6 +555,7 @@ impl ToolParser for DeepSeekDsmlParser { } else { self.buffer.clear(); } + self.completed_tool_call_count += 1; self.current_tool_id += 1; self.current_tool_name_sent = false; continue; @@ -500,25 +564,56 @@ impl ToolParser for DeepSeekDsmlParser { } } + // Once the outer close tag is complete, discard all DSML framing and + // release only genuine trailing assistant text. In live V4 streams the + // closing sentinel is one token followed by the block-name pieces, so + // treating only the opening sentinel as DSML leaks the close tag. + if let Some(start) = self.buffer.find(self.block_close.as_str()) { + let tail = self.buffer[start + self.block_close.len()..].replace(EOS_TOKEN, ""); + self.buffer.clear(); + normal_text.push_str(&tail); + } + Ok(StreamingParseResult { - normal_text: String::new(), + normal_text, calls: all_calls, }) } fn has_tool_markers(&self, text: &str) -> bool { - text.contains(self.block_open.as_str()) + text.contains(DSML_OPEN_SENTINEL) || text.contains(DSML_CLOSE_SENTINEL) } fn get_unstreamed_tool_args(&self) -> Option> { helpers::get_unstreamed_args(&self.prev_tool_call_arr, &self.streamed_args_for_tool) } + fn completed_tool_call_count(&self) -> Option { + Some(self.completed_tool_call_count) + } + + fn finish_incremental(&mut self) -> StreamingParseResult { + let marker_start = [DSML_OPEN_SENTINEL, DSML_CLOSE_SENTINEL] + .into_iter() + .filter_map(|marker| self.buffer.find(marker)) + .min(); + let normal_text = match marker_start { + Some(start) => self.buffer[..start].trim_end().to_string(), + None => std::mem::take(&mut self.buffer), + }; + self.buffer.clear(); + StreamingParseResult { + normal_text, + calls: Vec::new(), + } + } + fn reset(&mut self) { self.buffer.clear(); self.prev_tool_call_arr.clear(); self.current_tool_id = -1; self.current_tool_name_sent = false; self.streamed_args_for_tool.clear(); + self.completed_tool_call_count = 0; } } diff --git a/crates/tool_parser/src/traits.rs b/crates/tool_parser/src/traits.rs index 30ed8e8e0f..cb3894b758 100644 --- a/crates/tool_parser/src/traits.rs +++ b/crates/tool_parser/src/traits.rs @@ -45,6 +45,20 @@ pub trait ToolParser: Send + Sync { None } + /// Number of structurally complete, valid calls observed by an incremental + /// parser. `None` means the parser cannot distinguish completion from + /// provisional deltas and preserves the legacy finish-reason behavior. + fn completed_tool_call_count(&self) -> Option { + None + } + + /// Finalize buffered incremental input at the Engine terminal boundary. + /// Parsers use this to release ordinary text that was retained only to + /// disambiguate a split structural marker. + fn finish_incremental(&mut self) -> StreamingParseResult { + StreamingParseResult::default() + } + /// Reset the parser state for reuse across requests. /// This should clear all buffers and reset state to initial values. fn reset(&mut self) { diff --git a/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs b/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs index 5b549f1fdf..288bf2bef5 100644 --- a/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs +++ b/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs @@ -371,6 +371,27 @@ async fn test_deepseek_v4_complete_mixed_types() { assert_eq!(args["enabled"], true); } +#[tokio::test] +async fn test_deepseek_v4_complete_with_tools_omits_invalid_calls() { + let parser = DeepSeekDsmlParser::v4(); + let tools = create_test_tools(); + let input = concat!( + "Visible answer.\n\n", + "<｜DSML｜tool_calls>\n", + "<｜DSML｜invoke name=\"not_supplied\">{\"value\":1}\n", + "<｜DSML｜invoke name=\"get_weather\">{not-json}\n", + "", + "Ordinary trailing answer.", + ); + + let (normal_text, calls) = parser + .parse_complete_with_tools(input, &tools) + .await + .unwrap(); + assert_eq!(normal_text, "Visible answer.Ordinary trailing answer."); + assert!(calls.is_empty()); +} + #[test] fn test_deepseek_v4_format_detection() { let parser = DeepSeekDsmlParser::v4(); @@ -378,22 +399,23 @@ fn test_deepseek_v4_format_detection() { assert!(parser.has_tool_markers("<｜DSML｜tool_calls>")); assert!(parser.has_tool_markers("text <｜DSML｜tool_calls> marker")); - // V4 parser must NOT fire on the V3.2 block name. - assert!(!parser.has_tool_markers("<｜DSML｜function_calls>")); + // Any DSML sentinel is structural and must not leak as ordinary content, + // even when the block name is malformed for the selected model family. + assert!(parser.has_tool_markers("<｜DSML｜function_calls>")); assert!(!parser.has_tool_markers("plain text")); } #[test] fn test_deepseek_v32_does_not_match_v4_block() { - // Guardrail: a V3.2 parser must NOT treat a V4-shaped payload as a tool call. + // A cross-variant sentinel is structural even though it cannot form a call. let parser = DeepSeekDsmlParser::v32(); - assert!(!parser.has_tool_markers("<｜DSML｜tool_calls>")); + assert!(parser.has_tool_markers("<｜DSML｜tool_calls>")); } #[tokio::test] -async fn test_deepseek_v4_cross_variant_payload_passthrough() { - // A V4 parser given a V3.2-shaped payload must not parse calls; the input - // should flow through as normal text (has_tool_markers returns false). +async fn test_deepseek_v4_cross_variant_payload_is_not_a_call() { + // A V4 parser given a V3.2-shaped payload must not parse calls or expose + // the foreign DSML framing as normal content. let parser = DeepSeekDsmlParser::v4(); let v32_input = concat!( "<｜DSML｜function_calls>\n", @@ -404,7 +426,7 @@ async fn test_deepseek_v4_cross_variant_payload_passthrough() { ); let (normal_text, tools) = parser.parse_complete(v32_input).await.unwrap(); assert!(tools.is_empty(), "V4 parser must not parse V3.2 block"); - assert_eq!(normal_text, v32_input); + assert!(normal_text.is_empty()); } #[tokio::test] @@ -440,6 +462,7 @@ async fn test_deepseek_v4_streaming_single_tool() { assert!(found_name, "Should have found tool name during streaming"); assert!(!collected_args.is_empty(), "Should have streamed arguments"); + assert_eq!(parser.completed_tool_call_count(), Some(1)); } #[tokio::test] @@ -531,6 +554,57 @@ async fn test_deepseek_dsml_v4_streaming_strips_eos_from_partial_parameter() { !collected_args.contains("<｜end▁of▁sentence｜>"), "EOS must not leak into V4 streamed argument bytes, got: {collected_args:?}" ); + assert_eq!(parser.completed_tool_call_count(), Some(0)); +} + +#[tokio::test] +async fn test_deepseek_dsml_v4_terminal_invalid_call_is_not_complete() { + let tools = create_test_tools(); + let mut parser = DeepSeekDsmlParser::v4(); + let result = parser + .parse_incremental( + concat!( + "Visible answer.\n\n", + "<｜DSML｜tool_calls>", + "<｜DSML｜invoke name=\"get_weather\">{not-json}", + "", + "Ordinary trailing answer.", + ), + &tools, + ) + .await + .unwrap(); + + assert_eq!( + result.normal_text, + "Visible answer.Ordinary trailing answer." + ); + assert!(result.calls.is_empty()); + assert_eq!(parser.completed_tool_call_count(), Some(0)); +} + +#[tokio::test] +async fn test_deepseek_dsml_v4_terminal_partial_preserves_only_prefix_text() { + let tools = create_test_tools(); + let mut parser = DeepSeekDsmlParser::v4(); + let result = parser + .parse_incremental( + concat!( + "Visible answer.\n\n", + "<｜DSML｜tool_calls>", + "<｜DSML｜invoke name=\"get_weather\">", + "<｜DSML｜parameter name=\"city\" string=\"true\">Ber", + ), + &tools, + ) + .await + .unwrap(); + assert_eq!(result.normal_text, "Visible answer."); + assert_eq!(parser.completed_tool_call_count(), Some(0)); + + let terminal = parser.finish_incremental(); + assert!(terminal.normal_text.is_empty()); + assert!(terminal.calls.is_empty()); } /// A malformed complete invoke with `name=""` must not stall the buffer. @@ -674,8 +748,8 @@ async fn test_deepseek_dsml_v4_streaming_bpe_chunked_opener() { "argument bytes must be emitted, got: {collected_args:?}" ); assert!( - !normal_text.contains("<｜DSML｜"), - "DSML sentinel must not leak into normal_text, got: {normal_text:?}" + normal_text.is_empty(), + "DSML framing and its leading whitespace must not leak into normal_text, got: {normal_text:?}" ); } diff --git a/model_gateway/src/routers/grpc/regular/streaming.rs b/model_gateway/src/routers/grpc/regular/streaming.rs index 71427e68bc..16c9153b29 100644 --- a/model_gateway/src/routers/grpc/regular/streaming.rs +++ b/model_gateway/src/routers/grpc/regular/streaming.rs @@ -635,36 +635,57 @@ impl StreamingProcessor { // Phase 3: Check unstreamed tool args for (index, parser) in &tool_parsers { - let parser_guard = parser.lock().await; - if let Some(unstreamed_items) = parser_guard.get_unstreamed_tool_args() { - for tool_call_item in unstreamed_items { - let tool_call_delta = ToolCallDelta { - index: tool_call_item.tool_index as u32, - id: None, - tool_type: None, - function: Some(FunctionCallDelta { - name: None, - arguments: if tool_call_item.parameters.is_empty() { - None - } else { - Some(tool_call_item.parameters) - }, - }), - }; + let mut parser_guard = parser.lock().await; + let finalized = parser_guard.finish_incremental(); + if !finalized.normal_text.is_empty() { + let content_chunk = ChatCompletionStreamResponse::builder(request_id, model) + .created(created) + .add_choice_content(*index, "assistant", finalized.normal_text) + .maybe_system_fingerprint(system_fingerprint) + .build(); + let sse_chunk = sse_encoder + .encode_data(&content_chunk) + .map_err(|e| format!("Failed to serialize terminal content chunk: {e}"))?; + tx.send(Ok(sse_chunk)) + .await + .map_err(|_| "Failed to send terminal content chunk".to_string())?; + } + let terminal_items = finalized.calls.into_iter().chain( + parser_guard + .get_unstreamed_tool_args() + .into_iter() + .flatten(), + ); + for tool_call_item in terminal_items { + let tool_call_delta = ToolCallDelta { + index: tool_call_item.tool_index as u32, + id: None, + tool_type: None, + function: Some(FunctionCallDelta { + name: None, + arguments: if tool_call_item.parameters.is_empty() { + None + } else { + Some(tool_call_item.parameters) + }, + }), + }; - let tool_chunk = ChatCompletionStreamResponse::builder(request_id, model) - .created(created) - .add_choice_tool_call_delta(*index, tool_call_delta) - .maybe_system_fingerprint(system_fingerprint) - .build(); + let tool_chunk = ChatCompletionStreamResponse::builder(request_id, model) + .created(created) + .add_choice_tool_call_delta(*index, tool_call_delta) + .maybe_system_fingerprint(system_fingerprint) + .build(); - let sse_chunk = sse_encoder - .encode_data(&tool_chunk) - .map_err(|e| format!("Failed to serialize tool chunk: {e}"))?; - tx.send(Ok(sse_chunk)) - .await - .map_err(|_| "Failed to send unstreamed tool args".to_string())?; - } + let sse_chunk = sse_encoder + .encode_data(&tool_chunk) + .map_err(|e| format!("Failed to serialize tool chunk: {e}"))?; + tx.send(Ok(sse_chunk)) + .await + .map_err(|_| "Failed to send unstreamed tool args".to_string())?; + } + if let Some(completed_call_count) = parser_guard.completed_tool_call_count() { + has_tool_calls.insert(*index, completed_call_count > 0); } } From b500ce0250164892a7acfa558c51b1ae108308bf Mon Sep 17 00:00:00 2001 From: Zihua Wu <13583761+lucifer1004@users.noreply.github.com> Date: Thu, 20 Aug 2026 04:35:59 -0700 Subject: [PATCH 2/3] fix(tool-parser): finalize terminal tool calls in the messages stream path Mirror the chat path's terminal finalization in process_messages_streaming_chunks: flush retained normal text via finish_incremental() as a text content block, chain finalized calls with get_unstreamed_tool_args(), and reconcile has_tool_calls with completed_tool_call_count() so an all-invalid DSML stream degrades stop_reason from tool_use to end_turn instead of leaving a complete- looking tool_use block whose input JSON does not parse. Add a regression test covering a split invoke that streams provisional deltas, then closes with invalid JSON arguments and stays uncompleted. Signed-off-by: Zihua Wu <13583761+lucifer1004@users.noreply.github.com> --- .../tests/tool_parser_deepseek_dsml.rs | 47 ++++++ .../src/routers/grpc/regular/streaming.rs | 153 ++++++++++++------ 2 files changed, 150 insertions(+), 50 deletions(-) diff --git a/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs b/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs index 288bf2bef5..be852f990d 100644 --- a/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs +++ b/crates/tool_parser/tests/tool_parser_deepseek_dsml.rs @@ -607,6 +607,53 @@ async fn test_deepseek_dsml_v4_terminal_partial_preserves_only_prefix_text() { assert!(terminal.calls.is_empty()); } +/// Regression for the router terminal-flush finding: an invoke that streams +/// provisional name/argument deltas while still incomplete, then closes with +/// invalid JSON arguments, must not count as a completed call. The streaming +/// routers rely on `completed_tool_call_count()` staying zero here to degrade +/// the terminal finish/stop reason instead of pinning "tool_calls"/"tool_use". +#[tokio::test] +async fn test_deepseek_dsml_v4_split_invalid_invoke_stays_uncompleted() { + let tools = create_test_tools(); + let mut parser = DeepSeekDsmlParser::v4(); + + // Chunk 1: incomplete invoke — provisional deltas stream before the + // arguments can be validated. + let first = parser + .parse_incremental( + "<｜DSML｜tool_calls><｜DSML｜invoke name=\"get_weather\">{\"city\": \"Ber", + &tools, + ) + .await + .unwrap(); + assert!( + first + .calls + .iter() + .any(|call| call.name.as_deref() == Some("get_weather")), + "provisional tool name delta should stream before validation, got: {:?}", + first.calls + ); + assert_eq!(parser.completed_tool_call_count(), Some(0)); + + // Chunk 2 closes the invoke with invalid JSON arguments; the parser + // skips the call, so the completed count stays zero and only genuine + // trailing text is released. + let second = parser + .parse_incremental( + "lin\" oops}Trailing answer.", + &tools, + ) + .await + .unwrap(); + assert_eq!(second.normal_text, "Trailing answer."); + assert_eq!(parser.completed_tool_call_count(), Some(0)); + + let terminal = parser.finish_incremental(); + assert!(terminal.normal_text.is_empty()); + assert!(terminal.calls.is_empty()); +} + /// A malformed complete invoke with `name=""` must not stall the buffer. /// Previously the streaming `invoke_regex` required `[^"]+` so `name=""` /// never matched, leaving the bad block stuck at the head of the buffer — diff --git a/model_gateway/src/routers/grpc/regular/streaming.rs b/model_gateway/src/routers/grpc/regular/streaming.rs index 16c9153b29..63d0dc09c1 100644 --- a/model_gateway/src/routers/grpc/regular/streaming.rs +++ b/model_gateway/src/routers/grpc/regular/streaming.rs @@ -2388,74 +2388,127 @@ impl StreamingProcessor { } } - // Phase 3: Flush unstreamed tool args from the incremental parser - if let Some(ref parser) = streaming_tool_parser { - if let Some(unstreamed_items) = parser.get_unstreamed_tool_args() { - for tool_call_item in unstreamed_items { - has_tool_calls = true; - - if let Some(ref name) = tool_call_item.name { - // Close text block if open before starting tool block - if text_block_open { - Self::send_messages_event( - tx, - &mut sse_buffer, - &MessageStreamEvent::ContentBlockStop { - index: current_block_index, - }, - ) - .await?; - text_block_open = false; - current_block_index += 1; - } - if tool_block_open { - Self::send_messages_event( - tx, - &mut sse_buffer, - &MessageStreamEvent::ContentBlockStop { - index: current_block_index, - }, - ) - .await?; - current_block_index += 1; - } + // Phase 3: Finalize the incremental parser — flush held-back text, + // emit finalized/unstreamed tool args, then reconcile has_tool_calls + // with the count of completed valid calls so an all-invalid stream + // does not pin stop_reason to "tool_use". + if let Some(ref mut parser) = streaming_tool_parser { + let finalized = parser.finish_incremental(); + if !finalized.normal_text.is_empty() { + if tool_block_open { + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockStop { + index: current_block_index, + }, + ) + .await?; + tool_block_open = false; + current_block_index += 1; + } + if !text_block_open { + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockStart { + index: current_block_index, + content_block: ContentBlock::Text { + text: String::new(), + citations: None, + }, + }, + ) + .await?; + text_block_open = true; + } + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockDelta { + index: current_block_index, + delta: ContentBlockDelta::TextDelta { + text: finalized.normal_text, + }, + }, + ) + .await?; + } - let tool_call_id = utils::generate_tool_call_id( - model, - name, - tool_call_item.tool_index, - history_tool_calls_count, - ); + let terminal_items = finalized.calls.into_iter().chain( + parser.get_unstreamed_tool_args().into_iter().flatten(), + ); + for tool_call_item in terminal_items { + has_tool_calls = true; + + if let Some(ref name) = tool_call_item.name { + // Close text block if open before starting tool block + if text_block_open { Self::send_messages_event( tx, &mut sse_buffer, - &MessageStreamEvent::ContentBlockStart { + &MessageStreamEvent::ContentBlockStop { index: current_block_index, - content_block: ContentBlock::ToolUse { - id: message_utils::anthropic_tool_use_id(&tool_call_id), - name: name.clone(), - input: Value::Object(serde_json::Map::new()), - }, }, ) .await?; - tool_block_open = true; + text_block_open = false; + current_block_index += 1; } - - if !tool_call_item.parameters.is_empty() { + if tool_block_open { Self::send_messages_event( tx, &mut sse_buffer, - &MessageStreamEvent::ContentBlockDelta { + &MessageStreamEvent::ContentBlockStop { index: current_block_index, - delta: ContentBlockDelta::InputJsonDelta { - partial_json: tool_call_item.parameters, - }, }, ) .await?; + current_block_index += 1; } + + let tool_call_id = utils::generate_tool_call_id( + model, + name, + tool_call_item.tool_index, + history_tool_calls_count, + ); + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockStart { + index: current_block_index, + content_block: ContentBlock::ToolUse { + id: message_utils::anthropic_tool_use_id(&tool_call_id), + name: name.clone(), + input: Value::Object(serde_json::Map::new()), + }, + }, + ) + .await?; + tool_block_open = true; } + + if !tool_call_item.parameters.is_empty() { + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockDelta { + index: current_block_index, + delta: ContentBlockDelta::InputJsonDelta { + partial_json: tool_call_item.parameters, + }, + }, + ) + .await?; + } + } + + // Provisional deltas from a complete-but-invalid call were + // skipped by the parser; if no call actually completed, degrade + // the terminal stop_reason from "tool_use" to "end_turn". + if let Some(completed_call_count) = parser.completed_tool_call_count() { + has_tool_calls = completed_call_count > 0; } } From f3c928082d2e71d932c669016cc29e8ee222a8b7 Mon Sep 17 00:00:00 2001 From: Zihua Wu <13583761+lucifer1004@users.noreply.github.com> Date: Thu, 20 Aug 2026 06:30:37 -0700 Subject: [PATCH 3/3] fix(tool-parser): flush terminal tool args before terminal text in messages stream Emitting finalized text first closed a still-open tool_use block, so trailing InputJsonDelta items landed inside a text block and the tool call's JSON stayed truncated. Flush terminal tool-call items first, then the retained text. Signed-off-by: Zihua Wu <13583761+lucifer1004@users.noreply.github.com> --- .../src/routers/grpc/regular/streaming.rs | 85 ++++++++++--------- 1 file changed, 45 insertions(+), 40 deletions(-) diff --git a/model_gateway/src/routers/grpc/regular/streaming.rs b/model_gateway/src/routers/grpc/regular/streaming.rs index 63d0dc09c1..ac55a78563 100644 --- a/model_gateway/src/routers/grpc/regular/streaming.rs +++ b/model_gateway/src/routers/grpc/regular/streaming.rs @@ -2394,47 +2394,11 @@ impl StreamingProcessor { // does not pin stop_reason to "tool_use". if let Some(ref mut parser) = streaming_tool_parser { let finalized = parser.finish_incremental(); - if !finalized.normal_text.is_empty() { - if tool_block_open { - Self::send_messages_event( - tx, - &mut sse_buffer, - &MessageStreamEvent::ContentBlockStop { - index: current_block_index, - }, - ) - .await?; - tool_block_open = false; - current_block_index += 1; - } - if !text_block_open { - Self::send_messages_event( - tx, - &mut sse_buffer, - &MessageStreamEvent::ContentBlockStart { - index: current_block_index, - content_block: ContentBlock::Text { - text: String::new(), - citations: None, - }, - }, - ) - .await?; - text_block_open = true; - } - Self::send_messages_event( - tx, - &mut sse_buffer, - &MessageStreamEvent::ContentBlockDelta { - index: current_block_index, - delta: ContentBlockDelta::TextDelta { - text: finalized.normal_text, - }, - }, - ) - .await?; - } + // Flush terminal tool-call items first, while a tool_use block is + // still open: emitting text first would close the block and send + // trailing InputJsonDelta inside a text block, which the + // content-block contract forbids. let terminal_items = finalized.calls.into_iter().chain( parser.get_unstreamed_tool_args().into_iter().flatten(), ); @@ -2504,6 +2468,47 @@ impl StreamingProcessor { } } + if !finalized.normal_text.is_empty() { + if tool_block_open { + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockStop { + index: current_block_index, + }, + ) + .await?; + tool_block_open = false; + current_block_index += 1; + } + if !text_block_open { + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockStart { + index: current_block_index, + content_block: ContentBlock::Text { + text: String::new(), + citations: None, + }, + }, + ) + .await?; + text_block_open = true; + } + Self::send_messages_event( + tx, + &mut sse_buffer, + &MessageStreamEvent::ContentBlockDelta { + index: current_block_index, + delta: ContentBlockDelta::TextDelta { + text: finalized.normal_text, + }, + }, + ) + .await?; + } + // Provisional deltas from a complete-but-invalid call were // skipped by the parser; if no call actually completed, degrade // the terminal stop_reason from "tool_use" to "end_turn".