refactor(harmony): clean up parser signatures, fix delta accumulation, and GptOss tests

e2e_test/chat_completions/test_openai_server.py

@@ -12,7 +12,7 @@
 
 import pytest
 from conftest import smg_compare
-from infra import is_trtllm
+from infra import is_sglang, is_trtllm
 
 logger = logging.getLogger(__name__)
 
@@ -28,6 +28,10 @@
 class TestChatCompletion:
     """Tests for OpenAI-compatible chat completions API."""
 
+    # Whether the backend trims stop sequences from output.
+    # Harmony (gpt-oss) does not trim because its detokenization is not channel-aware.
+    STOP_SEQUENCE_TRIMMED = True
+
     @pytest.mark.parametrize("logprobs", [None, 5])
     @pytest.mark.parametrize("parallel_sample_num", [1, 2])
     def test_chat_completion(self, setup_backend, smg, logprobs, parallel_sample_num):
@@ -278,7 +282,8 @@ def test_streaming_token_count_matches_chunks(self, setup_backend, smg):
             f"content chunk count ({content_chunk_count})"
         )
         # But they should be close - allow small difference for special tokens
-        assert usage_completion_tokens - content_chunk_count <= 2, (
+        token_tolerance = getattr(self, "STREAMING_TOKEN_TOLERANCE", 2)
+        assert usage_completion_tokens - content_chunk_count <= token_tolerance, (
             f"completion_tokens ({usage_completion_tokens}) differs too much from "
             f"content chunk count ({content_chunk_count})"
         )
@@ -348,8 +353,14 @@ def test_stop_sequences(self, setup_backend, smg):
         )
 
         assert response.choices[0].finish_reason == "stop"
-        content = response.choices[0].message.content
-        assert "," not in content, f"Stop sequence ',' should not appear in output: {content}"
+        msg = response.choices[0].message
+        content = msg.content or getattr(msg, "reasoning_content", "") or ""
+        if self.STOP_SEQUENCE_TRIMMED:
+            assert "," not in content, f"Stop sequence ',' should not appear in output: {content}"
+        else:
+            assert content.endswith(","), (
+                f"Stop sequence ',' should be the suffix of output: {content}"
+            )
 
         # SmgClient comparison
         with smg_compare():
@@ -366,10 +377,16 @@ def test_stop_sequences(self, setup_backend, smg):
                 stop=[","],
             )
             assert smg_resp.choices[0].finish_reason == "stop"
-            smg_content = smg_resp.choices[0].message.content
-            assert "," not in smg_content, (
-                f"SmgClient: stop sequence ',' should not appear: {smg_content}"
-            )
+            smg_msg = smg_resp.choices[0].message
+            smg_content = smg_msg.content or getattr(smg_msg, "reasoning_content", "") or ""
+            if self.STOP_SEQUENCE_TRIMMED:
+                assert "," not in smg_content, (
+                    f"SmgClient: stop sequence ',' should not appear: {smg_content}"
+                )
+            else:
+                assert smg_content.endswith(","), (
+                    f"SmgClient: stop sequence ',' should be the suffix: {smg_content}"
+                )
 
     def test_stop_sequences_stream(self, setup_backend, smg):
         """Test that stop sequences work in streaming mode."""
@@ -397,11 +414,18 @@ def test_stop_sequences_stream(self, setup_backend, smg):
         ]
         assert "stop" in finish_reasons
 
-        # Collect all content
+        # Collect all content (fall back to reasoning_content for models like Harmony)
         content = "".join(
-            c.choices[0].delta.content for c in chunks if c.choices and c.choices[0].delta.content
+            self._delta_text(c.choices[0].delta)
+            for c in chunks
+            if c.choices and self._delta_text(c.choices[0].delta)
         )
-        assert "," not in content, f"Stop sequence ',' should not appear in output: {content}"
+        if self.STOP_SEQUENCE_TRIMMED:
+            assert "," not in content, f"Stop sequence ',' should not appear in output: {content}"
+        else:
+            assert content.endswith(","), (
+                f"Stop sequence ',' should be the suffix of output: {content}"
+            )
 
         # SmgClient streaming comparison
         with smg_compare():
@@ -426,16 +450,24 @@ def test_stop_sequences_stream(self, setup_backend, smg):
             ]
             assert "stop" in smg_finish
             smg_text = "".join(
-                c.choices[0].delta.content
+                self._delta_text(c.choices[0].delta)
                 for c in smg_chunks
-                if c.choices and c.choices[0].delta.content
+                if c.choices and self._delta_text(c.choices[0].delta)
             )
-            assert "," not in smg_text
+            if self.STOP_SEQUENCE_TRIMMED:
+                assert "," not in smg_text
+            else:
+                assert smg_text.endswith(",")
 
     # -------------------------------------------------------------------------
     # Helper methods
     # -------------------------------------------------------------------------
 
+    @staticmethod
+    def _delta_text(delta):
+        """Extract text from delta, falling back to reasoning_content for Harmony."""
+        return delta.content or getattr(delta, "reasoning_content", "") or ""
+
     def _run_chat_completion(self, client, model, logprobs, parallel_sample_num):
         """Run a non-streaming chat completion and verify response."""
         response = client.chat.completions.create(
@@ -558,6 +590,12 @@ class TestChatCompletionGptOss(TestChatCompletion):
     with OSS models. Logprobs are supported via Harmony's built-in tokenizer.
     """
 
+    # Harmony channel markers add ~10 special tokens
+    STREAMING_TOKEN_TOLERANCE = 10
+
+    # Harmony doesn't trim stop sequences (detokenization is not channel-aware)
+    STOP_SEQUENCE_TRIMMED = False
+
     @pytest.mark.parametrize("logprobs", [None, 5])
     @pytest.mark.parametrize("parallel_sample_num", [1, 2])
     def test_chat_completion(self, setup_backend, smg, logprobs, parallel_sample_num):
@@ -570,6 +608,18 @@ def test_chat_completion_stream(self, setup_backend, smg, logprobs, parallel_sam
         """Test streaming chat completion with logprobs and parallel sampling."""
         super().test_chat_completion_stream(setup_backend, smg, logprobs, parallel_sample_num)
 
+    def test_stop_sequences(self, setup_backend, smg):
+        if is_trtllm():
+            pytest.skip("TRT-LLM Harmony stop_word_ids path has known bugs")
+        super().test_stop_sequences(setup_backend, smg)
+
+    def test_stop_sequences_stream(self, setup_backend, smg):
+        if is_trtllm():
+            pytest.skip("TRT-LLM Harmony stop_word_ids path has known bugs")
+        if is_sglang():
+            self.STOP_SEQUENCE_TRIMMED = True
+        super().test_stop_sequences_stream(setup_backend, smg)
+
     @pytest.mark.skip(reason="OSS models don't support regex constraints")
     def test_regex(self, setup_backend, smg):
         pass
@@ -581,17 +631,3 @@ def test_penalty(self, setup_backend, smg):
     @pytest.mark.skip(reason="OSS models don't support continue_final_message")
     def test_response_prefill(self, setup_backend, smg):
         pass
-
-    @pytest.mark.skip(reason="TODO: Harmony puts output in reasoning_content, content is None")
-    def test_stop_sequences(self, setup_backend):
-        pass
-
-    @pytest.mark.skip(reason="TODO: Harmony puts output in reasoning_content, content is None")
-    def test_stop_sequences_stream(self, setup_backend):
-        pass
-
-    @pytest.mark.skip(
-        reason="TODO: Harmony channel markers add ~10 tokens exceeding parent tolerance of 2"
-    )
-    def test_streaming_token_count_matches_chunks(self, setup_backend):
-        pass

model_gateway/src/routers/grpc/harmony/parser.rs

@@ -226,21 +226,10 @@ impl HarmonyParserAdapter {
     ///
     /// Parses all output token IDs and returns the complete channel output
     /// containing analysis, commentary (tool calls), and final text.
-    ///
-    /// # Arguments
-    ///
-    /// * `output_ids` - The complete output token IDs from the model
-    /// * `finish_reason` - The finish reason from GenerateComplete ("stop", "length", etc.)
-    /// * `matched_stop` - Optional matched stop token information from GenerateComplete
-    ///
-    /// # Returns
-    ///
-    /// Complete HarmonyChannelOutput with all three channels parsed
     pub fn parse_complete(
         &mut self,
         output_ids: &[u32],
         finish_reason: String,
-        matched_stop: Option<serde_json::Value>,
     ) -> Result<HarmonyChannelOutput, String> {
         let mut reasoning_token_count = 0u32;
 
@@ -279,7 +268,6 @@ impl HarmonyParserAdapter {
             commentary,
             final_text,
             finish_reason: final_finish_reason,
-            matched_stop,
             reasoning_token_count,
         })
     }
@@ -356,10 +344,9 @@ impl HarmonyParserAdapter {
         &mut self,
         chunk_ids: &[u32],
     ) -> Result<Option<HarmonyChannelDelta>, String> {
-        let mut has_delta = false;
-        let mut analysis_delta = None;
+        let mut analysis_delta: Option<String> = None;
         let mut commentary_delta = None;
-        let mut final_delta = None;
+        let mut final_delta: Option<String> = None;
 
         // Track message count before processing
         let prev_message_count = self.parser.messages().len();
@@ -382,16 +369,18 @@ impl HarmonyParserAdapter {
 
             // Check for content delta
             if let Ok(Some(delta_text)) = self.parser.last_content_delta() {
-                has_delta = true;
-
                 // Determine which channel this delta belongs to
                 let channel = self.parser.current_channel();
                 match channel.as_deref() {
                     Some("analysis") => {
-                        analysis_delta = Some(delta_text);
+                        analysis_delta
+                            .get_or_insert_with(String::new)
+                            .push_str(&delta_text);
                     }
                     Some("final") | None => {
-                        final_delta = Some(delta_text);
+                        final_delta
+                            .get_or_insert_with(String::new)
+                            .push_str(&delta_text);
                     }
                     Some("commentary") => {
                         // Accumulate delta for commentary
@@ -406,8 +395,6 @@ impl HarmonyParserAdapter {
         if self.parser.current_channel().as_deref() == Some("commentary") {
             if let Some(cur_recipient) = self.parser.current_recipient() {
                 if let Some(tool_name) = cur_recipient.strip_prefix("functions.") {
-                    has_delta = true;
-
                     // Count completed tool calls for index
                     let base_index = self
                         .parser
@@ -459,6 +446,10 @@ impl HarmonyParserAdapter {
         let current_message_count = self.parser.messages().len();
         let is_final = current_message_count > prev_message_count;
 
+        // Only emit a delta if at least one channel produced output
+        let has_delta =
+            analysis_delta.is_some() || commentary_delta.is_some() || final_delta.is_some();
+
         if has_delta {
             Ok(Some(HarmonyChannelDelta {
                 analysis_delta,
@@ -479,16 +470,10 @@ impl HarmonyParserAdapter {
     /// # Arguments
     ///
     /// * `finish_reason` - The finish reason from GenerateComplete ("stop", "length", etc.)
-    /// * `matched_stop` - Optional matched stop token information from GenerateComplete
-    ///
     /// # Returns
     ///
     /// Final HarmonyChannelOutput with complete parsed content
-    pub fn finalize(
-        &mut self,
-        finish_reason: String,
-        matched_stop: Option<serde_json::Value>,
-    ) -> HarmonyChannelOutput {
+    pub fn finalize(&mut self, finish_reason: String) -> HarmonyChannelOutput {
         // Extract all completed messages
         let messages = self.parser.messages();
 
@@ -510,7 +495,6 @@ impl HarmonyParserAdapter {
             commentary,
             final_text,
             finish_reason: final_finish_reason,
-            matched_stop,
             reasoning_token_count: self.reasoning_token_count,
         }
     }

model_gateway/src/routers/grpc/harmony/processor.rs

@@ -74,13 +74,9 @@ impl HarmonyResponseProcessor {
                 )
             })?;
 
-            // Parse Harmony channels with finish_reason and matched_stop
+            // Parse Harmony channels with finish_reason
             let parsed = parser
-                .parse_complete(
-                    complete.output_ids(),
-                    complete.finish_reason().to_string(),
-                    matched_stop.clone(),
-                )
+                .parse_complete(complete.output_ids(), complete.finish_reason().to_string())
                 .map_err(|e| {
                     error!(
                         function = "process_non_streaming_chat_response",
@@ -219,14 +215,8 @@ impl HarmonyResponseProcessor {
             )
         })?;
 
-        let matched_stop = complete.matched_stop_json();
-
         let parsed = parser
-            .parse_complete(
-                complete.output_ids(),
-                complete.finish_reason().to_string(),
-                matched_stop,
-            )
+            .parse_complete(complete.output_ids(), complete.finish_reason().to_string())
             .map_err(|e| {
                 error!(
                     function = "process_responses_iteration",

model_gateway/src/routers/grpc/harmony/streaming.rs

@@ -283,15 +283,14 @@ impl HarmonyStreamingProcessor {
                     // Finalize parser and emit final chunk
                     if let Some(parser) = parsers.get_mut(&index) {
                         let matched_stop = matched_stops.get(&index).and_then(|m| m.clone());
-                        let final_output = parser.finalize(
-                            complete_wrapper.finish_reason().to_string(),
-                            matched_stop.clone(),
-                        );
+
+                        let final_output =
+                            parser.finalize(complete_wrapper.finish_reason().to_string());
 
                         Self::emit_final_chunk(
                             index,
                             &final_output.finish_reason,
-                            final_output.matched_stop.as_ref(),
+                            matched_stop.as_ref(),
                             dispatch,
                             original_request,
                             tx,
@@ -547,8 +546,8 @@ impl HarmonyStreamingProcessor {
             HashMap::new();
 
         // Metadata from Complete message; seed cached_tokens from prefill phase (dual-stream)
-        let mut finish_reason = String::from("stop");
-        let mut matched_stop: Option<serde_json::Value> = None;
+        let mut finish_reason: String;
+        let mut finalized_analysis: Option<String> = None;
         let mut prompt_tokens: u32 = 0;
         let mut completion_tokens: u32 = 0;
         let mut cached_tokens: u32 = prefill_cached_tokens;
@@ -784,7 +783,6 @@ impl HarmonyStreamingProcessor {
                 ProtoResponseVariant::Complete(complete_wrapper) => {
                     // Store final metadata
                     finish_reason = complete_wrapper.finish_reason().to_string();
-                    matched_stop = complete_wrapper.matched_stop_json();
                     prompt_tokens = complete_wrapper.prompt_tokens();
                     // Combine decode-stream cached_tokens with any prefill cached_tokens
                     cached_tokens = cached_tokens.saturating_add(complete_wrapper.cached_tokens());
@@ -795,10 +793,12 @@ impl HarmonyStreamingProcessor {
                     }
 
                     // Finalize parser and get complete output
-                    let final_output = parser.finalize(finish_reason.clone(), matched_stop.clone());
+                    // Responses API: no user-specified stop sequences
+                    let final_output = parser.finalize(finish_reason.clone());
 
-                    // Store finalized tool calls and reasoning token count
-                    accumulated_tool_calls.clone_from(&final_output.commentary);
+                    // Store finalized output for later use
+                    finalized_analysis = final_output.analysis;
+                    accumulated_tool_calls = final_output.commentary;
                     reasoning_token_count = final_output.reasoning_token_count;
 
                     // Complete all tool calls if we have commentary
@@ -1011,10 +1011,7 @@ impl HarmonyStreamingProcessor {
         if let Some(tool_calls) = accumulated_tool_calls {
             if !tool_calls.is_empty() {
                 let analysis_content = if has_analysis {
-                    // Get analysis from finalized parser output by calling finalize again
-                    // This is safe because finalize can be called multiple times
-                    let output = parser.finalize(finish_reason.clone(), matched_stop.clone());
-                    output.analysis
+                    finalized_analysis
                 } else {
                     None
                 };

model_gateway/src/routers/grpc/harmony/types.rs

@@ -3,7 +3,6 @@
 use openai_harmony::chat::Content;
 use openai_protocol::common::ToolCall;
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
 
 /// Harmony message format
 ///
@@ -99,9 +98,6 @@ pub(crate) struct HarmonyChannelOutput {
     /// Finish reason
     pub finish_reason: String,
 
-    /// Matched stop token (if any)
-    pub matched_stop: Option<Value>,
-
     /// Number of reasoning tokens (from analysis and commentary channels)
     pub reasoning_token_count: u32,
 }