dynamo-parsers 1.3.0

// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

use std::fmt::Debug;

use crate::ParserResult;
use crate::ReasoningParser;

use openai_harmony::StreamableParser;
use openai_harmony::chat::{Content, Message, TextContent};
use openai_harmony::{HarmonyEncoding, HarmonyEncodingName, chat::Role, load_harmony_encoding};
use regex::Regex;

///// Static initialization of harmony encoder to not affect performance every time a parser is created
/// This is because load_harmony_encoding downloads some tiktoken files into a directory and we don't want to do this every time we create a parser.
use std::sync::OnceLock;

static GLOBAL_HARMONY_GPTOSS_ENCODING: OnceLock<Result<HarmonyEncoding, anyhow::Error>> =
    OnceLock::new();
static HARMONY_CALL_TOKEN_IDS: OnceLock<Vec<u32>> = OnceLock::new();
static HARMONY_ANALYSIS_BLOCK_REGEX: OnceLock<Regex> = OnceLock::new();
const HARMONY_CALL_MARKER: &str = "<|call|>";
const HARMONY_SPECIAL_TOKENS: &[&str] = &[
    "<|start|>",
    "<|end|>",
    "<|return|>",
    "<|channel|>",
    "<|constrain|>",
    "<|message|>",
    "<|call|>",
];

fn get_harmony_encoding() -> &'static Result<HarmonyEncoding, anyhow::Error> {
    GLOBAL_HARMONY_GPTOSS_ENCODING.get_or_init(|| {
        // load_harmony_encoding internally constructs a reqwest::blocking::Client,
        // which builds and drops a Tokio Runtime. Dropping a Runtime from inside
        // an async context (e.g. when this is called for the first time during
        // an HTTP request handler) panics with "Cannot drop a runtime in a
        // context where blocking is not allowed". Run the load on a fresh OS
        // thread so the inner Runtime is dropped outside any async context.
        // The init runs at most once per process via OnceLock.
        std::thread::spawn(|| load_harmony_encoding(HarmonyEncodingName::HarmonyGptOss))
            .join()
            .unwrap_or_else(|_| Err(anyhow::anyhow!("harmony encoding loader thread panicked")))
    })
}

pub struct GptOssReasoningParser {
    parser: StreamableParser,
    pending_text: String,
    pending_tool_call_text: String,
    emitted_reasoning_text: bool,
    insert_reasoning_separator: bool,
    emitted_normal_text: bool,
    insert_normal_separator: bool,
}

/// Implement Debug for GptOssReasoningParser separately because StreamableParser does not implement Debug
impl Debug for GptOssReasoningParser {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("GptOssReasoningParser")
            .field("parser", &self.parser.state_json())
            .finish()
    }
}

impl GptOssReasoningParser {
    pub fn new() -> anyhow::Result<Self> {
        let parser = match get_harmony_encoding().as_ref() {
            Ok(enc) => match StreamableParser::new(enc.clone(), Some(Role::Assistant)) {
                Ok(p) => p,
                Err(e) => {
                    tracing::warn!("Harmony StreamableParser init failed for GPT OSS: {e}");
                    return Err(anyhow::anyhow!(
                        "Failed to load Harmony StreamableParser: {e}"
                    ));
                }
            },
            Err(e) => {
                tracing::warn!("Failed to load Harmony encoding for GPT OSS: {e}");
                return Err(anyhow::anyhow!("Failed to load Harmony encoding: {e}"));
            }
        };
        Ok(Self {
            parser,
            pending_text: String::new(),
            pending_tool_call_text: String::new(),
            emitted_reasoning_text: false,
            insert_reasoning_separator: false,
            emitted_normal_text: false,
            insert_normal_separator: false,
        })
    }
}

fn encode_text_to_tokens(text: &str) -> anyhow::Result<Vec<u32>> {
    let enc = get_harmony_encoding()
        .as_ref()
        .map_err(|e| anyhow::anyhow!("Failed to get harmony encoding: {e}"))?;
    Ok(enc.tokenizer().encode_with_special_tokens(text))
}

fn harmony_call_token_ids() -> Option<&'static [u32]> {
    let ids = HARMONY_CALL_TOKEN_IDS.get_or_init(|| {
        get_harmony_encoding()
            .as_ref()
            .map(|enc| {
                enc.tokenizer()
                    .encode_with_special_tokens(HARMONY_CALL_MARKER)
            })
            .unwrap_or_default()
    });
    if ids.is_empty() {
        None
    } else {
        Some(ids.as_slice())
    }
}

/// Token id(s) for the harmony `<|call|>` terminator.
///
/// `<|call|>` is simultaneously the harmony tool-call terminator the parser
/// requires AND a gpt-oss EOS token. Pipelines that hide EOS tokens from the
/// decoded output (e.g. the preprocessor's stop-condition handling) must keep
/// these ids visible when the harmony tool-call parser is active — otherwise the
/// terminator is stripped before the parser sees it and tool calls are silently
/// dropped. Empty if the harmony encoding is unavailable.
pub fn harmony_terminator_token_ids() -> Vec<u32> {
    harmony_call_token_ids()
        .map(|ids| ids.to_vec())
        .unwrap_or_default()
}

fn token_ids_contain_sequence(token_ids: &[u32], marker_ids: &[u32]) -> bool {
    !marker_ids.is_empty()
        && marker_ids.len() <= token_ids.len()
        && token_ids
            .windows(marker_ids.len())
            .any(|window| window == marker_ids)
}

fn harmony_analysis_block_regex() -> &'static Regex {
    HARMONY_ANALYSIS_BLOCK_REGEX.get_or_init(|| {
        Regex::new(
            r"(?s)(?:<\|start\|>assistant)?<\|channel\|>analysis.*?<\|message\|>.*?(?:<\|call\|>|<\|end\|>|\z)",
        )
            .expect("harmony analysis block regex")
    })
}

fn input_contains_call_marker(
    text: &str,
    token_ids: &[u32],
    caller_supplied_token_ids: bool,
) -> bool {
    if !token_ids.is_empty()
        && let Some(call_ids) = harmony_call_token_ids()
    {
        return token_ids_contain_sequence(token_ids, call_ids);
    }

    !caller_supplied_token_ids && text.contains(HARMONY_CALL_MARKER)
}

fn raw_input_text_for_tool_parser(
    text: &str,
    token_ids: &[u32],
    caller_supplied_token_ids: bool,
) -> String {
    if caller_supplied_token_ids && let Ok(enc) = get_harmony_encoding() {
        return enc.tokenizer().decode_utf8(token_ids).unwrap_or_default();
    }

    text.to_string()
}

fn split_incomplete_harmony_suffix(text: &str) -> (&str, &str) {
    let hold_len = HARMONY_SPECIAL_TOKENS
        .iter()
        .flat_map(|token| (1..token.len()).map(|idx| &token[..idx]))
        .filter(|prefix| text.ends_with(*prefix))
        .map(str::len)
        .max()
        .unwrap_or(0);

    text.split_at(text.len() - hold_len)
}

fn contains_directed_harmony_function_call(text: &str) -> bool {
    text.contains("<|channel|>commentary to=functions.") && text.contains(HARMONY_CALL_MARKER)
}

fn strip_analysis_blocks_for_tool_handoff(text: &str) -> String {
    harmony_analysis_block_regex()
        .replace_all(text, "")
        .into_owned()
}

fn append_separated(target: &mut String, text: &str) {
    if text.is_empty() {
        return;
    }
    if !target.is_empty() {
        target.push('\n');
    }
    target.push_str(text);
}

fn append_text_content(target: &mut String, content: &[Content]) {
    for item in content {
        if let Content::Text(TextContent { text }) = item {
            append_separated(target, text);
        }
    }
}

fn append_message_by_channel(reasoning_text: &mut String, normal_text: &mut String, msg: &Message) {
    match msg.channel.as_deref() {
        Some("analysis") => append_text_content(reasoning_text, &msg.content),
        Some("final") => append_text_content(normal_text, &msg.content),
        Some("commentary") if msg.recipient.is_none() => {
            append_text_content(normal_text, &msg.content)
        }
        _ => {}
    }
}

fn append_current_by_channel(
    reasoning_text: &mut String,
    normal_text: &mut String,
    channel: Option<String>,
    current: String,
) {
    match channel.as_deref() {
        Some("analysis") => append_separated(reasoning_text, &current),
        Some("final") => append_separated(normal_text, &current),
        Some("commentary") => append_separated(normal_text, &current),
        _ => {}
    }
}

fn is_visible_normal_channel(channel: Option<&str>, recipient: Option<&str>) -> bool {
    matches!(channel, Some("final"))
        || (matches!(channel, Some("commentary")) && recipient.is_none())
}

fn starts_directed_harmony_tool_call(text: &str) -> bool {
    text.contains("<|channel|>commentary to=functions.")
}

impl ReasoningParser for GptOssReasoningParser {
    fn finish_reasoning_stream(&mut self) -> ParserResult {
        self.pending_tool_call_text.clear();
        let pending = std::mem::take(&mut self.pending_text);
        if pending.is_empty() {
            return ParserResult::default();
        }

        match self.parser.current_channel().as_deref() {
            Some("analysis") => ParserResult {
                normal_text: String::new(),
                reasoning_text: pending,
            },
            Some("final") | Some("commentary") => ParserResult {
                normal_text: pending,
                reasoning_text: String::new(),
            },
            _ => ParserResult::default(),
        }
    }

    fn detect_and_parse_reasoning(&mut self, text: &str, token_ids: &[u32]) -> ParserResult {
        let caller_supplied_token_ids = !token_ids.is_empty();
        let encoded_tokens;
        let token_ids = if token_ids.is_empty() {
            // WAR: Since we are moving to just text based reasoning parsing, converting to token_ids now using harmony encoding
            encoded_tokens = match encode_text_to_tokens(text) {
                Ok(tokens) => tokens,
                Err(err) => {
                    tracing::warn!("Failed to encode Harmony tokens: {err}");
                    return ParserResult::default();
                }
            };
            encoded_tokens.as_slice()
        } else {
            token_ids
        };

        let parser = &mut self.parser;

        for (i, token_id) in token_ids.iter().enumerate() {
            tracing::debug!(
                "Processing token {} of {}: {}",
                i + 1,
                token_ids.len(),
                token_id
            );
            if let Err(e) = parser.process(*token_id) {
                tracing::warn!("Harmony parse error for token_id {token_id}: {e}");
                return ParserResult::default();
            }
        }

        let output_msgs = parser.messages();
        tracing::debug!("Parser has {} output messages", output_msgs.len());

        let mut reasoning_text = String::new();
        let mut normal_text = String::new();
        for msg in output_msgs {
            append_message_by_channel(&mut reasoning_text, &mut normal_text, msg);
        }

        let current = parser.current_content().unwrap_or_default();
        let current_channel = parser.current_channel();
        if current_channel.as_deref() != Some("commentary") || parser.current_recipient().is_none()
        {
            append_current_by_channel(
                &mut reasoning_text,
                &mut normal_text,
                current_channel,
                current,
            );
        }

        let raw_input_text =
            raw_input_text_for_tool_parser(text, token_ids, caller_supplied_token_ids);
        if contains_directed_harmony_function_call(&raw_input_text) {
            // The serving pipeline feeds reasoning normal_text into the Harmony
            // tool parser. Keep directed commentary envelopes intact so tool
            // calls are not silently dropped.
            normal_text = strip_analysis_blocks_for_tool_handoff(&raw_input_text);
        }

        tracing::debug!(
            "Final result - normal_text: {} chars, reasoning_text: {} chars",
            normal_text.len(),
            reasoning_text.len()
        );

        ParserResult {
            normal_text,
            reasoning_text,
        }
    }

    fn parse_reasoning_streaming_incremental(
        &mut self,
        text: &str,
        token_ids: &[u32],
    ) -> ParserResult {
        let caller_supplied_token_ids = !token_ids.is_empty();
        let text_to_process;
        let text = if caller_supplied_token_ids {
            text
        } else {
            self.pending_text.push_str(text);
            let combined = std::mem::take(&mut self.pending_text);
            let (ready, pending) = split_incomplete_harmony_suffix(&combined);
            self.pending_text.push_str(pending);
            text_to_process = ready.to_string();
            text_to_process.as_str()
        };

        if text.is_empty() && token_ids.is_empty() {
            return ParserResult::default();
        }

        let encoded_tokens;
        let token_ids = if token_ids.is_empty() {
            // WAR: Since we are moving to just text based reasoning parsing, converting to token_ids now using harmony encoding
            encoded_tokens = match encode_text_to_tokens(text) {
                Ok(tokens) => tokens,
                Err(err) => {
                    tracing::warn!("Failed to encode Harmony tokens: {err}");
                    return ParserResult::default();
                }
            };
            encoded_tokens.as_slice()
        } else {
            token_ids
        };

        let parser: &mut StreamableParser = &mut self.parser;
        let mut normal_delta = String::new();
        let mut reasoning_delta = String::new();

        for (i, token_id) in token_ids.iter().enumerate() {
            tracing::debug!(
                "Processing streaming token {} of {}: {}",
                i + 1,
                token_ids.len(),
                token_id
            );
            let previous_channel = parser.current_channel();
            let previous_recipient = parser.current_recipient();
            if let Err(e) = parser.process(*token_id) {
                tracing::warn!("Harmony parse error for token_id {token_id}: {e}");
                return ParserResult::default();
            }
            let current_channel = parser.current_channel();
            let current_recipient = parser.current_recipient();

            if previous_channel.as_deref() != Some("analysis")
                && current_channel.as_deref() == Some("analysis")
                && self.emitted_reasoning_text
            {
                self.insert_reasoning_separator = true;
            }

            if is_visible_normal_channel(current_channel.as_deref(), current_recipient.as_deref())
                && !is_visible_normal_channel(
                    previous_channel.as_deref(),
                    previous_recipient.as_deref(),
                )
                && self.emitted_normal_text
            {
                self.insert_normal_separator = true;
            }

            if let (Some(delta), Some(channel)) = (
                parser.last_content_delta().unwrap_or_default(),
                current_channel,
            ) {
                // `last_content_delta` only exposes the newest token slice, so directed
                // commentary still falls through to the raw handoff path for tool parsing.
                match channel.as_str() {
                    "final" => {
                        if self.insert_normal_separator {
                            normal_delta.push('\n');
                            self.insert_normal_separator = false;
                        }
                        normal_delta.push_str(&delta);
                        self.emitted_normal_text = true;
                    }
                    "analysis" => {
                        if self.insert_reasoning_separator {
                            reasoning_delta.push('\n');
                            self.insert_reasoning_separator = false;
                        }
                        reasoning_delta.push_str(&delta);
                        self.emitted_reasoning_text = true;
                    }
                    "commentary" => {
                        if current_recipient.is_none() {
                            if self.insert_normal_separator {
                                normal_delta.push('\n');
                                self.insert_normal_separator = false;
                            }
                            normal_delta.push_str(&delta);
                            self.emitted_normal_text = true;
                        }
                    }
                    _ => {}
                }
            }
        }

        let has_call_marker =
            input_contains_call_marker(text, token_ids, caller_supplied_token_ids);
        let raw_input_text = if has_call_marker
            || !self.pending_tool_call_text.is_empty()
            || starts_directed_harmony_tool_call(text)
        {
            Some(raw_input_text_for_tool_parser(
                text,
                token_ids,
                caller_supplied_token_ids,
            ))
        } else {
            None
        };

        if let Some(raw_input_text) = raw_input_text {
            // Streaming currently feeds the downstream Harmony tool parser through
            // `normal_text`. This is an internal handoff, not visible-content
            // semantics: directed commentary tool payloads must become tool calls,
            // not assistant `content`.
            if !self.pending_tool_call_text.is_empty() {
                self.pending_tool_call_text.push_str(&raw_input_text);
                if has_call_marker {
                    normal_delta.push_str(&std::mem::take(&mut self.pending_tool_call_text));
                }
            } else if starts_directed_harmony_tool_call(&raw_input_text) && !has_call_marker {
                self.pending_tool_call_text.push_str(&raw_input_text);
            } else if has_call_marker {
                normal_delta.push_str(&raw_input_text);
            }
        }

        if !normal_delta.is_empty() || !reasoning_delta.is_empty() {
            tracing::debug!(
                "Returning aggregated deltas: normal: {} chars, reasoning: {} chars",
                normal_delta.len(),
                reasoning_delta.len()
            );
            return ParserResult {
                normal_text: normal_delta,
                reasoning_text: reasoning_delta,
            };
        }

        if let Some(channel) = parser.current_channel() {
            if channel == "commentary" {
                tracing::debug!("In commentary channel, recovering full content");
                // If we're in the commentary channel, we should return raw token content and recover content that has been consumed by the parser
                // so that the tool parser can process it properly
                if let Ok(enc) = get_harmony_encoding() {
                    let current_content = parser.current_content().unwrap_or_default();
                    let mut final_text = text.to_string();

                    // Restore commentary metadata consumed by the parser so the tool-call parser can
                    // process it correctly.
                    //
                    // Example:
                    //   Before parsing:
                    //   "<|start|>assistant<|channel|>commentary to=functions.get_current_weather <|constrain|>json<|message|>{\"format\":\"celsius\",\"location\":\"San Francisco\"}<|call|>"
                    //   After parsing, the header is stripped, so we must reconstruct it:
                    //   "<|channel|>commentary to=functions.get_current_weather <|constrain|>json<|message|>"
                    //
                    // This ensures downstream tool-call parsing receives the channel, target, and
                    // constraint metadata together with the message payload.

                    // Recovery should only happen once, and only when `current_content` is empty.
                    if current_content.is_empty() {
                        let tokens = parser.tokens();

                        // Get the token id for " <|channel|>"
                        let channel_token_id = enc
                            .tokenizer()
                            .encode_with_special_tokens("<|channel|>")
                            .last()
                            .copied();

                        // Find the last occurrence of the <|channel|> token (id 20005) in the tokens vector
                        let last_channel_token_idx = channel_token_id
                            .and_then(|token_id| {
                                tokens.iter().rposition(|token| *token == token_id)
                            })
                            .unwrap_or(0);

                        // Then get the generated text from the last <|channel|> to the end of parser.tokens()
                        let end_token_idx = parser.tokens().len();
                        // Use Harmony's decode_utf8 to decode tokens into text
                        let generated_text = enc
                            .tokenizer()
                            .decode_utf8(&parser.tokens()[last_channel_token_idx..end_token_idx])
                            .unwrap_or_default();

                        final_text = generated_text;
                    }

                    return ParserResult {
                        normal_text: final_text,
                        reasoning_text: String::new(),
                    };
                }
            } else {
                tracing::warn!("Shouldn't be delta content after in channel: {}", channel);
            }
        }
        tracing::debug!("No deltas to return, returning empty result");
        ParserResult::default()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test] // REASONING.batch.2.c, TOOLCALLING.harmony.1
    fn test_gpt_oss_reasoning_parser() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let text = "<|channel|>analysis<|message|>The user asks a simple factual question: capital of Brazil. The answer is Brasília. No additional explanation needed.<|end|><|start|>assistant<|channel|>final<|message|>The capital of Brazil is Brasília.";
        let result = parser.detect_and_parse_reasoning(text, &[]);
        assert!(result.normal_text == "The capital of Brazil is Brasília.");
        assert!(
            result.reasoning_text
                == "The user asks a simple factual question: capital of Brazil. The answer is Brasília. No additional explanation needed."
        );
    }

    #[test] // REASONING.batch.4
    fn test_gpt_oss_final_channel_without_analysis_is_normal_text() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let result = parser.detect_and_parse_reasoning("<|channel|>final<|message|>answer", &[]);
        assert_eq!(result.reasoning_text, "");
        assert_eq!(result.normal_text, "answer");
    }

    #[test] // REASONING.batch.6.a
    fn test_gpt_oss_multiple_analysis_spans_join_before_final() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let text = "<|channel|>analysis<|message|>first<|end|><|start|>assistant<|channel|>analysis<|message|>second<|end|><|start|>assistant<|channel|>final<|message|>done";
        let result = parser.detect_and_parse_reasoning(text, &[]);
        assert_eq!(result.reasoning_text, "first\nsecond");
        assert_eq!(result.normal_text, "done");
    }

    #[test]
    fn test_gpt_oss_directed_commentary_is_tool_parser_handoff() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let text = "<|channel|>analysis<|message|>think<|end|><|start|>assistant<|channel|>commentary to=functions.get_weather <|constrain|>json<|message|>{\"city\":\"SF\"}<|call|><|start|>assistant<|channel|>final<|message|>It is sunny.";
        let result = parser.detect_and_parse_reasoning(text, &[]);
        assert_eq!(result.reasoning_text, "think");
        assert_eq!(
            result.normal_text,
            "<|start|>assistant<|channel|>commentary to=functions.get_weather <|constrain|>json<|message|>{\"city\":\"SF\"}<|call|><|start|>assistant<|channel|>final<|message|>It is sunny."
        );
    }

    #[test]
    fn test_gpt_oss_analysis_call_does_not_swallow_commentary_handoff() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let text = "<|channel|>analysis to=functions.search <|constrain|>json<|message|>{\"q\":\"x\"}<|call|><|start|>assistant<|channel|>commentary to=functions.get_weather <|constrain|>json<|message|>{\"city\":\"SF\"}<|call|><|start|>assistant<|channel|>final<|message|>It is sunny.";
        let result = parser.detect_and_parse_reasoning(text, &[]);
        assert_eq!(result.reasoning_text, "{\"q\":\"x\"}");
        assert_eq!(
            result.normal_text,
            "<|start|>assistant<|channel|>commentary to=functions.get_weather <|constrain|>json<|message|>{\"city\":\"SF\"}<|call|><|start|>assistant<|channel|>final<|message|>It is sunny."
        );
    }

    #[test]
    fn test_gpt_oss_recipientless_commentary_is_normal_text() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let text = "<|channel|>commentary<|message|>I will check that.<|end|><|start|>assistant<|channel|>final<|message|>Done.";
        let result = parser.detect_and_parse_reasoning(text, &[]);
        assert_eq!(result.reasoning_text, "");
        assert_eq!(result.normal_text, "I will check that.\nDone.");
    }

    #[test] // REASONING.stream.2.a, REASONING.batch.2.c, TOOLCALLING.harmony.1
    fn test_gpt_oss_reasoning_parser_streaming() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let chunks = vec![
            "<|channel|>",
            "analysis<|message|>The user asks a simple factual question: capital of Brazil.",
            " The answer is Brasília. No additional explanation needed.",
            "<|end|><|start|>assistant<|channel|>final<|message|>",
            "The capital of Brazil is Brasília.",
        ];
        let mut reasoning_text_incr = String::new();
        let mut normal_text_incr = String::new();
        for chunk in chunks {
            let result = parser.parse_reasoning_streaming_incremental(chunk, &[]);
            normal_text_incr.push_str(&result.normal_text);
            reasoning_text_incr.push_str(&result.reasoning_text);
        }
        assert!(normal_text_incr == "The capital of Brazil is Brasília.");
        assert!(
            reasoning_text_incr
                == "The user asks a simple factual question: capital of Brazil. The answer is Brasília. No additional explanation needed."
        );
    }

    #[test] // REASONING.stream.3.b
    fn test_gpt_oss_reasoning_parser_streaming_split_end_marker() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let chunks = vec![
            "<|channel|>analysis<|message|>thinking<|e",
            "nd|><|start|>assistant<|channel|>final<|message|>answer",
        ];
        let mut reasoning_text_incr = String::new();
        let mut normal_text_incr = String::new();
        for chunk in chunks {
            let result = parser.parse_reasoning_streaming_incremental(chunk, &[]);
            normal_text_incr.push_str(&result.normal_text);
            reasoning_text_incr.push_str(&result.reasoning_text);
        }
        assert_eq!(reasoning_text_incr, "thinking");
        assert_eq!(normal_text_incr, "answer");
    }

    #[test]
    fn test_gpt_oss_reasoning_parser_streaming_flushes_pending_suffix_on_finish() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let result = parser.parse_reasoning_streaming_incremental(
            "<|channel|>analysis<|message|>thinking<|e",
            &[],
        );
        let finish = parser.finish_reasoning_stream();
        assert_eq!(result.reasoning_text, "thinking");
        assert_eq!(finish.reasoning_text, "<|e");
        assert_eq!(finish.normal_text, "");
        let finish = parser.finish_reasoning_stream();
        assert_eq!(finish.reasoning_text, "");
        assert_eq!(finish.normal_text, "");

        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let result = parser
            .parse_reasoning_streaming_incremental("<|channel|>final<|message|>answer<|e", &[]);
        let finish = parser.finish_reasoning_stream();
        assert_eq!(result.normal_text, "answer");
        assert_eq!(finish.normal_text, "<|e");
        assert_eq!(finish.reasoning_text, "");
    }

    #[test] // REASONING.stream.2.b
    fn test_gpt_oss_reasoning_parser_streaming_multiple_analysis_spans() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let chunks = vec![
            "<|channel|>analysis<|message|>first<|end|><|start|>assistant",
            "<|channel|>analysis<|message|>second<|end|><|start|>assistant",
            "<|channel|>final<|message|>done",
        ];
        let mut reasoning_text_incr = String::new();
        let mut normal_text_incr = String::new();
        for chunk in chunks {
            let result = parser.parse_reasoning_streaming_incremental(chunk, &[]);
            normal_text_incr.push_str(&result.normal_text);
            reasoning_text_incr.push_str(&result.reasoning_text);
        }
        assert_eq!(reasoning_text_incr, "first\nsecond");
        assert_eq!(normal_text_incr, "done");
    }

    #[test]
    fn test_gpt_oss_reasoning_parser_streaming_recipientless_commentary() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let chunks = vec![
            "<|channel|>commentary<|message|>I will check that.<|end|><|start|>assistant",
            "<|channel|>final<|message|>Done.",
        ];
        let mut reasoning_text_incr = String::new();
        let mut normal_text_incr = String::new();
        for chunk in chunks {
            let result = parser.parse_reasoning_streaming_incremental(chunk, &[]);
            normal_text_incr.push_str(&result.normal_text);
            reasoning_text_incr.push_str(&result.reasoning_text);
        }
        assert_eq!(reasoning_text_incr, "");
        assert_eq!(normal_text_incr, "I will check that.\nDone.");
    }

    #[test]
    fn test_gpt_oss_reasoning_parser_streaming_split_directed_commentary() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let chunks = vec![
            "<|channel|>commentary to=functions.get_",
            "weather <|constrain|>json<|message|>{\"city\":\"SF\"}<|call|>",
        ];
        let mut reasoning_text_incr = String::new();
        let mut normal_text_incr = String::new();
        for chunk in chunks {
            let result = parser.parse_reasoning_streaming_incremental(chunk, &[]);
            normal_text_incr.push_str(&result.normal_text);
            reasoning_text_incr.push_str(&result.reasoning_text);
        }
        assert_eq!(reasoning_text_incr, "");
        assert_eq!(
            normal_text_incr,
            "<|channel|>commentary to=functions.get_weather <|constrain|>json<|message|>{\"city\":\"SF\"}<|call|>"
        );
    }

    #[test] // REASONING.stream.2.a, REASONING.batch.2.c, TOOLCALLING.harmony.1
    fn test_gpt_oss_reasoning_parser_streaming_chunked() {
        let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
        let enc = get_harmony_encoding()
            .as_ref()
            .expect("Failed to get encoding");
        let text = "<|channel|>analysis<|message|>The user asks a simple factual question: capital of Brazil. The answer is Brasília. No additional explanation needed.<|end|><|start|>assistant<|channel|>final<|message|>The capital of Brazil is Brasília.";
        let token_ids = enc.tokenizer().encode_with_special_tokens(text);
        let mut reasoning_text_incr = String::new();
        let mut normal_text_incr = String::new();

        let mut idx = 0;
        let chunk_size = 4;
        while idx < token_ids.len() {
            let end = (idx + chunk_size).min(token_ids.len());
            let result =
                parser.parse_reasoning_streaming_incremental("Test text", &token_ids[idx..end]);
            normal_text_incr.push_str(&result.normal_text);
            reasoning_text_incr.push_str(&result.reasoning_text);
            idx = end;
        }

        assert_eq!(normal_text_incr, "The capital of Brazil is Brasília.");
        assert_eq!(
            reasoning_text_incr,
            "The user asks a simple factual question: capital of Brazil. The answer is Brasília. No additional explanation needed."
        );
    }

    #[test] // REASONING.batch.3.a, TOOLCALLING.harmony.1
    fn test_gpt_oss_reasoning_parser_streaming_variable_length_chunks() {
        let text = "<|channel|>analysis<|message|>User asks: \"Hey, quick check: is everything up and running?\" We should check system health using the provided function get_system_health. Use function.<|end|><|start|>assistant<|channel|>commentary to=functions.get_system_health <|constrain|>json<|message|>{}";
        let enc = get_harmony_encoding()
            .as_ref()
            .expect("Failed to get encoding");
        let token_ids = enc.tokenizer().encode_with_special_tokens(text);

        // Send token one by one
        {
            let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
            let mut reasoning_text_incr = String::new();
            let mut normal_text_incr = String::new();
            for token in token_ids.iter() {
                let result = parser.parse_reasoning_streaming_incremental("", &[(*token)]);
                normal_text_incr.push_str(&result.normal_text);
                reasoning_text_incr.push_str(&result.reasoning_text);
            }
            assert_eq!(
                reasoning_text_incr,
                "User asks: \"Hey, quick check: is everything up and running?\" We should check system health using the provided function get_system_health. Use function."
            );
            // [gluo TODO] missing "<|start|>assistant" and "{}" from original message
            assert_eq!(
                normal_text_incr,
                "<|channel|>commentary to=functions.get_system_health <|constrain|>json<|message|>"
            );
        }

        // Send token in chunks (chunking obtained from actual model output)
        {
            let mut parser = GptOssReasoningParser::new().expect("Failed to create parser");
            let mut reasoning_text_incr = String::new();
            let mut normal_text_incr = String::new();
            let chunk_tokens = [
                vec![200005],
                vec![35644, 200008, 1844, 31064, 25, 392, 25216, 11, 4853],
                vec![2371, 25, 382, 5519, 869, 326, 6788, 16842, 1416, 1757],
                vec![2371, 2420, 3230, 2360, 290, 5181, 1114, 717, 39303, 126214],
                vec![
                    13, 7649, 1114, 13, 200007, 200006, 173781, 200005, 12606, 815,
                ],
                vec![
                    316, 28, 44580, 775, 39303, 126214, 220, 200003, 4108, 200008,
                ],
                vec![12083],
            ];
            // Concatenate chunk tokens and verify they match original token_ids
            let concatenated: Vec<u32> = chunk_tokens.iter().flatten().copied().collect();
            assert_eq!(concatenated, token_ids);

            for token in chunk_tokens.iter() {
                let result = parser.parse_reasoning_streaming_incremental("", token);
                normal_text_incr.push_str(&result.normal_text);
                reasoning_text_incr.push_str(&result.reasoning_text);
            }
            assert_eq!(
                reasoning_text_incr,
                "User asks: \"Hey, quick check: is everything up and running?\" We should check system health using the provided function get_system_health. Use function."
            );
            assert_eq!(
                normal_text_incr,
                "<|channel|>commentary to=functions.get_system_health <|constrain|>json<|message|>"
            );
        }
    }
}